From b5dec3df393e2d569547f3816bb326561162e095 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Thu, 18 Dec 2025 15:27:49 +0100 Subject: [PATCH] Track network serial in engine --- client/internal/engine.go | 2 +- shared/management/client/client.go | 2 +- shared/management/client/grpc.go | 41 ++++++------------------------ shared/management/client/mock.go | 6 ++--- 4 files changed, 13 insertions(+), 38 deletions(-) diff --git a/client/internal/engine.go b/client/internal/engine.go index 16ffdb33a..6ad8e1101 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -972,7 +972,7 @@ func (e *Engine) receiveManagementEvents() { e.config.LazyConnectionEnabled, ) - err = e.mgmClient.Sync(e.ctx, info, e.handleSync) + err = e.mgmClient.Sync(e.ctx, info, e.networkSerial, e.handleSync) if err != nil { // happens if management is unavailable for a long time. // We want to cancel the operation of the whole client diff --git a/shared/management/client/client.go b/shared/management/client/client.go index 3126bcd1f..2cafd7657 100644 --- a/shared/management/client/client.go +++ b/shared/management/client/client.go @@ -13,7 +13,7 @@ import ( type Client interface { io.Closer - Sync(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error + Sync(ctx context.Context, sysInfo *system.Info, networkSerial uint64, msgHandler func(msg *proto.SyncResponse) error) error GetServerPublicKey() (*wgtypes.Key, error) Register(serverKey wgtypes.Key, setupKey string, jwtToken string, sysInfo *system.Info, sshKey []byte, dnsLabels domain.List) (*proto.LoginResponse, error) Login(serverKey wgtypes.Key, sysInfo *system.Info, sshKey []byte, dnsLabels domain.List) (*proto.LoginResponse, error) diff --git a/shared/management/client/grpc.go b/shared/management/client/grpc.go index 2d6f4dbf9..cbb74ce3d 100644 --- a/shared/management/client/grpc.go +++ b/shared/management/client/grpc.go @@ -44,9 +44,6 @@ type GrpcClient struct { conn *grpc.ClientConn connStateCallback ConnStateNotifier connStateCallbackLock sync.RWMutex - // lastNetworkMapSerial stores last seen network map serial to optimize sync - lastNetworkMapSerial uint64 - lastNetworkMapSerialMu sync.Mutex } // NewClient creates a new client to Management service @@ -113,7 +110,7 @@ func (c *GrpcClient) ready() bool { // Sync wraps the real client's Sync endpoint call and takes care of retries and encryption/decryption of messages // Blocking request. The result will be sent via msgHandler callback function -func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error { +func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, networkSerial uint64, msgHandler func(msg *proto.SyncResponse) error) error { operation := func() error { log.Debugf("management connection state %v", c.conn.GetState()) connState := c.conn.GetState() @@ -131,7 +128,7 @@ func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler return err } - return c.handleStream(ctx, *serverPubKey, sysInfo, msgHandler) + return c.handleStream(ctx, *serverPubKey, sysInfo, networkSerial, msgHandler) } err := backoff.Retry(operation, defaultBackoff(ctx)) @@ -143,11 +140,11 @@ func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler } func (c *GrpcClient) handleStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info, - msgHandler func(msg *proto.SyncResponse) error) error { + networkSerial uint64, msgHandler func(msg *proto.SyncResponse) error) error { ctx, cancelStream := context.WithCancel(ctx) defer cancelStream() - stream, err := c.connectToStream(ctx, serverPubKey, sysInfo) + stream, err := c.connectToStream(ctx, serverPubKey, sysInfo, networkSerial) if err != nil { log.Debugf("failed to open Management Service stream: %s", err) if s, ok := gstatus.FromError(err); ok && s.Code() == codes.PermissionDenied { @@ -189,7 +186,8 @@ func (c *GrpcClient) GetNetworkMap(sysInfo *system.Info) (*proto.NetworkMap, err ctx, cancelStream := context.WithCancel(c.ctx) defer cancelStream() - stream, err := c.connectToStream(ctx, *serverPubKey, sysInfo) + // GetNetworkMap doesn't have a serial to send, so we pass 0 + stream, err := c.connectToStream(ctx, *serverPubKey, sysInfo, 0) if err != nil { log.Debugf("failed to open Management Service stream: %s", err) return nil, err @@ -219,13 +217,10 @@ func (c *GrpcClient) GetNetworkMap(sysInfo *system.Info) (*proto.NetworkMap, err return nil, fmt.Errorf("invalid msg, required network map") } - // update last seen serial - c.setLastNetworkMapSerial(decryptedResp.GetNetworkMap().GetSerial()) - return decryptedResp.GetNetworkMap(), nil } -func (c *GrpcClient) connectToStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info) (proto.ManagementService_SyncClient, error) { +func (c *GrpcClient) connectToStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info, networkSerial uint64) (proto.ManagementService_SyncClient, error) { // Always compute latest system info to ensure up-to-date PeerSystemMeta on first and subsequent syncs recomputed := system.GetInfo(c.ctx) if sysInfo != nil { @@ -235,7 +230,7 @@ func (c *GrpcClient) connectToStream(ctx context.Context, serverPubKey wgtypes.K recomputed.Files = sysInfo.Files } } - req := &proto.SyncRequest{Meta: infoToMetaData(recomputed), NetworkMapSerial: c.getLastNetworkMapSerial()} + req := &proto.SyncRequest{Meta: infoToMetaData(recomputed), NetworkMapSerial: networkSerial} myPrivateKey := c.key myPublicKey := myPrivateKey.PublicKey() @@ -273,11 +268,6 @@ func (c *GrpcClient) receiveEvents(stream proto.ManagementService_SyncClient, se return err } - // track latest network map serial if present - if decryptedResp.GetNetworkMap() != nil { - c.setLastNetworkMapSerial(decryptedResp.GetNetworkMap().GetSerial()) - } - if err := msgHandler(decryptedResp); err != nil { log.Errorf("failed handling an update message received from Management Service: %v", err.Error()) } @@ -602,18 +592,3 @@ func infoToMetaData(info *system.Info) *proto.PeerSystemMeta { }, } } - -// setLastNetworkMapSerial updates the cached last seen network map serial in a 32-bit safe manner -func (c *GrpcClient) setLastNetworkMapSerial(serial uint64) { - c.lastNetworkMapSerialMu.Lock() - c.lastNetworkMapSerial = serial - c.lastNetworkMapSerialMu.Unlock() -} - -// getLastNetworkMapSerial returns the cached last seen network map serial in a 32-bit safe manner -func (c *GrpcClient) getLastNetworkMapSerial() uint64 { - c.lastNetworkMapSerialMu.Lock() - v := c.lastNetworkMapSerial - c.lastNetworkMapSerialMu.Unlock() - return v -} diff --git a/shared/management/client/mock.go b/shared/management/client/mock.go index 29006c9c3..8b7f17ec3 100644 --- a/shared/management/client/mock.go +++ b/shared/management/client/mock.go @@ -12,7 +12,7 @@ import ( type MockClient struct { CloseFunc func() error - SyncFunc func(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error + SyncFunc func(ctx context.Context, sysInfo *system.Info, networkSerial uint64, msgHandler func(msg *proto.SyncResponse) error) error GetServerPublicKeyFunc func() (*wgtypes.Key, error) RegisterFunc func(serverKey wgtypes.Key, setupKey string, jwtToken string, info *system.Info, sshKey []byte, dnsLabels domain.List) (*proto.LoginResponse, error) LoginFunc func(serverKey wgtypes.Key, info *system.Info, sshKey []byte, dnsLabels domain.List) (*proto.LoginResponse, error) @@ -33,11 +33,11 @@ func (m *MockClient) Close() error { return m.CloseFunc() } -func (m *MockClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error { +func (m *MockClient) Sync(ctx context.Context, sysInfo *system.Info, networkSerial uint64, msgHandler func(msg *proto.SyncResponse) error) error { if m.SyncFunc == nil { return nil } - return m.SyncFunc(ctx, sysInfo, msgHandler) + return m.SyncFunc(ctx, sysInfo, networkSerial, msgHandler) } func (m *MockClient) GetServerPublicKey() (*wgtypes.Key, error) {