sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-10-26 14:29:08 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix concurrency and context

Browse Source
This commit is contained in:
pascal
2026-06-09 20:46:05 +02:00
parent bcfaa82d0a
commit ecf925f847
6 changed files with 20 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
management/server/affected_peers_test.go
+4
View File
@@ -1433,6 +1433,10 @@ func TestAffectedPeers_IsolatedEntitiesOnlyAffectTheirPeers(t *testing.T) {
updateManager.CloseChannel(ctx, peer3.ID) updateManager.CloseChannel(ctx, peer3.ID)
}) })
// The setup policy/route above dispatch affected-peer updates asynchronously;
// drain any in-flight ones so the assertions only observe the UpdateGroup below.
settleAffectedUpdates(updMsg1, updMsg2, updMsg3)
t.Run("policy group change does not affect route-only peer", func(t *testing.T) { t.Run("policy group change does not affect route-only peer", func(t *testing.T) {
done := make(chan struct{}) done := make(chan struct{})
go func() { go func() {
management/server/group.go
+2 -2
View File
@@ -269,7 +269,7 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us
storeEvent() storeEvent()
} }
am.dispatchAffected(ctx, accountID, snaps, changes) go am.dispatchAffected(ctx, accountID, snaps, changes)
return globalErr return globalErr
} }
@@ -312,7 +312,7 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us
storeEvent() storeEvent()
} }
am.dispatchAffected(ctx, accountID, snaps, changes) go am.dispatchAffected(ctx, accountID, snaps, changes)
return globalErr return globalErr
} }
management/server/networks/manager.go
+1 -1
View File
@@ -187,7 +187,7 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw
event() event()
} }
go m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change)
return nil return nil
} }
management/server/networks/resources/manager.go
+3 -3
View File
@@ -130,7 +130,7 @@ func (m *managerImpl) CreateResource(ctx context.Context, userID string, resourc
event() event()
} }
go m.accountManager.ExpandAndUpdateAffected(ctx, resource.AccountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, resource.AccountID, snap, change)
return resource, nil return resource, nil
} }
@@ -297,7 +297,7 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc
} }
}() }()
go m.accountManager.ExpandAndUpdateAffected(ctx, resource.AccountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, resource.AccountID, snap, change)
return resource, nil return resource, nil
} }
@@ -397,7 +397,7 @@ func (m *managerImpl) DeleteResource(ctx context.Context, accountID, userID, net
event() event()
} }
go m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change)
return nil return nil
} }
management/server/networks/routers/manager.go
+3 -3
View File
@@ -126,7 +126,7 @@ func (m *managerImpl) CreateRouter(ctx context.Context, userID string, router *t
m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterCreated, router.EventMeta(network)) m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterCreated, router.EventMeta(network))
go m.accountManager.ExpandAndUpdateAffected(ctx, router.AccountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, router.AccountID, snap, change)
return router, nil return router, nil
} }
@@ -175,7 +175,7 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t
m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterUpdated, router.EventMeta(network)) m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterUpdated, router.EventMeta(network))
go m.accountManager.ExpandAndUpdateAffected(ctx, router.AccountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, router.AccountID, snap, change)
return router, nil return router, nil
} }
@@ -256,7 +256,7 @@ func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, netwo
event() event()
go m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change) m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change)
return nil return nil
} }
management/server/peer.go
+7 -3
View File
@@ -1477,14 +1477,18 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
// never holds the write lock, over the consistent in-tx snapshot. Exported so the // never holds the write lock, over the consistent in-tx snapshot. Exported so the
// networks sub-package managers (which hold only account.Manager) share it. // networks sub-package managers (which hold only account.Manager) share it.
func (am *DefaultAccountManager) ExpandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) { func (am *DefaultAccountManager) ExpandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) {
am.dispatchAffected(ctx, accountID, []*affectedpeers.Snapshot{snap}, []affectedpeers.Change{change}) go am.dispatchAffected(ctx, accountID, []*affectedpeers.Snapshot{snap}, []affectedpeers.Change{change})
} }
// dispatchAffected expands one or more (snapshot, change) pairs — collected across // dispatchAffected expands one or more (snapshot, change) pairs — collected across
// one or several transactions — unions their affected peers, and dispatches a // one or several transactions — unions their affected peers, and dispatches a
// single network-map refresh. Each snapshot must already be loaded inside its // single network-map refresh. Each snapshot must already be loaded inside its
// transaction; this runs AFTER commit (pure in-memory + dispatch). // transaction; this runs AFTER commit (pure in-memory + dispatch). It is spawned
// in a goroutine that outlives the request, so it detaches from the request
// context's cancellation up front.
func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) { func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) {
ctx = context.WithoutCancel(ctx)
var lists [][]string var lists [][]string
for i, snap := range snaps { for i, snap := range snaps {
if snap == nil { if snap == nil {
@@ -1500,7 +1504,7 @@ func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID
} }
log.WithContext(ctx).Debugf("updating %d affected peers for account %s: %v", len(affectedPeerIDs), accountID, affectedPeerIDs) log.WithContext(ctx).Debugf("updating %d affected peers for account %s: %v", len(affectedPeerIDs), accountID, affectedPeerIDs)
_ = am.networkMapController.UpdateAffectedPeers(context.WithoutCancel(ctx), accountID, affectedPeerIDs) _ = am.networkMapController.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs)
} }
// unionStrings concatenates the given string lists into one deduplicated slice, // unionStrings concatenates the given string lists into one deduplicated slice,