Compare commits

...

5 Commits

Author SHA1 Message Date
Pascal Fischer
06c5564dd2 disable initial sync 2025-04-24 21:24:51 +02:00
Pascal Fischer
cace564a19 disable ephermeral manager and token manager 2025-04-24 21:08:45 +02:00
Pascal Fischer
d07c83c111 disable read locks 2025-04-24 20:59:59 +02:00
Pascal Fischer
b0a5696bec disable write locks 2025-04-24 20:48:22 +02:00
Pascal Fischer
dc24b9e276 make startup period configurable 2025-04-23 23:18:02 +02:00
3 changed files with 52 additions and 36 deletions

View File

@@ -196,14 +196,19 @@ func BuildManager(
}
var initialInterval int64
intervalStr := os.Getenv("PEER_UPDATE_INTERVAL_MS")
intervalStr := os.Getenv("NB_PEER_UPDATE_INTERVAL_MS")
interval, err := strconv.Atoi(intervalStr)
if err != nil {
initialInterval = 1
} else {
initialInterval = int64(interval) * 10
go func() {
time.Sleep(30 * time.Second)
startupPeriodStr := os.Getenv("NB_PEER_UPDATE_STARTUP_PERIOD_S")
startupPeriod, err := strconv.Atoi(startupPeriodStr)
if err != nil {
startupPeriod = 1
}
time.Sleep(time.Duration(startupPeriod) * time.Second)
am.updateAccountPeersBufferInterval.Store(int64(time.Duration(interval) * time.Millisecond))
log.WithContext(ctx).Infof("set peer update buffer interval to %dms", interval)
}()
@@ -1439,9 +1444,15 @@ func (am *DefaultAccountManager) SyncAndMarkPeer(ctx context.Context, accountID
peerUnlock := am.Store.AcquireWriteLockByUID(ctx, peerPubKey)
defer peerUnlock()
peer, netMap, postureChecks, err := am.SyncPeer(ctx, types.PeerSync{WireGuardPubKey: peerPubKey, Meta: meta}, accountID)
// peer, netMap, postureChecks, err := am.SyncPeer(ctx, types.PeerSync{WireGuardPubKey: peerPubKey, Meta: meta}, accountID)
// if err != nil {
// return nil, nil, nil, fmt.Errorf("error syncing peer: %w", err)
// }
peer, err := am.Store.GetPeerByPeerPubKey(ctx, store.LockingStrengthUpdate, peerPubKey)
if err != nil {
return nil, nil, nil, fmt.Errorf("error syncing peer: %w", err)
log.WithContext(ctx).Errorf("error getting peer by pubkey %s: %v", peerPubKey, err)
return nil, nil, nil, err
}
err = am.MarkPeerConnected(ctx, peerPubKey, true, realIP, accountID)
@@ -1449,7 +1460,7 @@ func (am *DefaultAccountManager) SyncAndMarkPeer(ctx context.Context, accountID
log.WithContext(ctx).Warnf("failed marking peer as connected %s %v", peerPubKey, err)
}
return peer, netMap, postureChecks, nil
return peer, &types.NetworkMap{}, []*posture.Checks{}, nil
}
func (am *DefaultAccountManager) OnPeerDisconnected(ctx context.Context, accountID string, peerPubKey string) error {

View File

@@ -172,23 +172,23 @@ func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementServi
log.WithContext(ctx).Tracef("peer system meta has to be provided on sync. Peer %s, remote addr %s", peerKey.String(), realIP)
}
peer, netMap, postureChecks, err := s.accountManager.SyncAndMarkPeer(ctx, accountID, peerKey.String(), extractPeerMeta(ctx, syncReq.GetMeta()), realIP)
peer, _, _, err := s.accountManager.SyncAndMarkPeer(ctx, accountID, peerKey.String(), extractPeerMeta(ctx, syncReq.GetMeta()), realIP)
if err != nil {
log.WithContext(ctx).Debugf("error while syncing peer %s: %v", peerKey.String(), err)
return mapError(ctx, err)
}
err = s.sendInitialSync(ctx, peerKey, peer, netMap, postureChecks, srv)
if err != nil {
log.WithContext(ctx).Debugf("error while sending initial sync for %s: %v", peerKey.String(), err)
return err
}
// err = s.sendInitialSync(ctx, peerKey, peer, netMap, postureChecks, srv)
// if err != nil {
// log.WithContext(ctx).Debugf("error while sending initial sync for %s: %v", peerKey.String(), err)
// return err
// }
updates := s.peersUpdateManager.CreateChannel(ctx, peer.ID)
s.ephemeralManager.OnPeerConnected(ctx, peer)
// s.ephemeralManager.OnPeerConnected(ctx, peer)
s.secretsManager.SetupRefresh(ctx, accountID, peer.ID)
// s.secretsManager.SetupRefresh(ctx, accountID, peer.ID)
if s.appMetrics != nil {
s.appMetrics.GRPCMetrics().CountSyncRequestDuration(time.Since(reqStart))

View File

@@ -133,36 +133,41 @@ func (s *SqlStore) AcquireGlobalLock(ctx context.Context) (unlock func()) {
// AcquireWriteLockByUID acquires an ID lock for writing to a resource and returns a function that releases the lock
func (s *SqlStore) AcquireWriteLockByUID(ctx context.Context, uniqueID string) (unlock func()) {
log.WithContext(ctx).Tracef("acquiring write lock for ID %s", uniqueID)
start := time.Now()
value, _ := s.resourceLocks.LoadOrStore(uniqueID, &sync.RWMutex{})
mtx := value.(*sync.RWMutex)
mtx.Lock()
unlock = func() {
mtx.Unlock()
log.WithContext(ctx).Tracef("released write lock for ID %s in %v", uniqueID, time.Since(start))
// log.WithContext(ctx).Tracef("acquiring write lock for ID %s", uniqueID)
//
// start := time.Now()
// value, _ := s.resourceLocks.LoadOrStore(uniqueID, &sync.RWMutex{})
// mtx := value.(*sync.RWMutex)
// mtx.Lock()
//
// unlock = func() {
// mtx.Unlock()
// log.WithContext(ctx).Tracef("released write lock for ID %s in %v", uniqueID, time.Since(start))
// }
//
// return unlock
return func() {
// noop
}
return unlock
}
// AcquireReadLockByUID acquires an ID lock for writing to a resource and returns a function that releases the lock
func (s *SqlStore) AcquireReadLockByUID(ctx context.Context, uniqueID string) (unlock func()) {
log.WithContext(ctx).Tracef("acquiring read lock for ID %s", uniqueID)
// log.WithContext(ctx).Tracef("acquiring read lock for ID %s", uniqueID)
//
// start := time.Now()
// value, _ := s.resourceLocks.LoadOrStore(uniqueID, &sync.RWMutex{})
// mtx := value.(*sync.RWMutex)
// mtx.RLock()
//
// unlock = func() {
// mtx.RUnlock()
// log.WithContext(ctx).Tracef("released read lock for ID %s in %v", uniqueID, time.Since(start))
// }
start := time.Now()
value, _ := s.resourceLocks.LoadOrStore(uniqueID, &sync.RWMutex{})
mtx := value.(*sync.RWMutex)
mtx.RLock()
unlock = func() {
mtx.RUnlock()
log.WithContext(ctx).Tracef("released read lock for ID %s in %v", uniqueID, time.Since(start))
return func() {
// noop
}
return unlock
}
func (s *SqlStore) SaveAccount(ctx context.Context, account *types.Account) error {