mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-15 09:59:08 +02:00
[management] remove lock and continue user update on failure (#4410)
This commit is contained in:
+51
-38
@@ -202,35 +202,45 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us
|
|||||||
}
|
}
|
||||||
|
|
||||||
var eventsToStore []func()
|
var eventsToStore []func()
|
||||||
var groupsToSave []*types.Group
|
|
||||||
var updateAccountPeers bool
|
var updateAccountPeers bool
|
||||||
|
|
||||||
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
|
var globalErr error
|
||||||
groupIDs := make([]string, 0, len(groups))
|
groupIDs := make([]string, 0, len(groups))
|
||||||
for _, newGroup := range groups {
|
for _, newGroup := range groups {
|
||||||
|
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
|
||||||
if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil {
|
if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
newGroup.AccountID = accountID
|
newGroup.AccountID = accountID
|
||||||
groupsToSave = append(groupsToSave, newGroup)
|
|
||||||
|
if err = transaction.CreateGroup(ctx, newGroup); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = transaction.IncrementNetworkSerial(ctx, accountID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
groupIDs = append(groupIDs, newGroup.ID)
|
groupIDs = append(groupIDs, newGroup.ID)
|
||||||
|
|
||||||
events := am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup)
|
events := am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup)
|
||||||
eventsToStore = append(eventsToStore, events...)
|
eventsToStore = append(eventsToStore, events...)
|
||||||
}
|
|
||||||
|
|
||||||
updateAccountPeers, err = areGroupChangesAffectPeers(ctx, transaction, accountID, groupIDs)
|
return nil
|
||||||
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
log.WithContext(ctx).Errorf("failed to update group %s: %v", newGroup.ID, err)
|
||||||
|
if len(groupIDs) == 1 {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
globalErr = errors.Join(globalErr, err)
|
||||||
|
// continue updating other groups
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if err = transaction.CreateGroups(ctx, accountID, groupsToSave); err != nil {
|
updateAccountPeers, err = areGroupChangesAffectPeers(ctx, am.Store, accountID, groupIDs)
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
return transaction.IncrementNetworkSerial(ctx, accountID)
|
|
||||||
})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -243,7 +253,7 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us
|
|||||||
am.UpdateAccountPeers(ctx, accountID)
|
am.UpdateAccountPeers(ctx, accountID)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return globalErr
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateGroups updates groups in the account.
|
// UpdateGroups updates groups in the account.
|
||||||
@@ -260,35 +270,45 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us
|
|||||||
}
|
}
|
||||||
|
|
||||||
var eventsToStore []func()
|
var eventsToStore []func()
|
||||||
var groupsToSave []*types.Group
|
|
||||||
var updateAccountPeers bool
|
var updateAccountPeers bool
|
||||||
|
|
||||||
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
|
var globalErr error
|
||||||
groupIDs := make([]string, 0, len(groups))
|
groupIDs := make([]string, 0, len(groups))
|
||||||
for _, newGroup := range groups {
|
for _, newGroup := range groups {
|
||||||
|
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
|
||||||
if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil {
|
if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
newGroup.AccountID = accountID
|
newGroup.AccountID = accountID
|
||||||
groupsToSave = append(groupsToSave, newGroup)
|
|
||||||
groupIDs = append(groupIDs, newGroup.ID)
|
if err = transaction.UpdateGroup(ctx, newGroup); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = transaction.IncrementNetworkSerial(ctx, accountID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
events := am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup)
|
events := am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup)
|
||||||
eventsToStore = append(eventsToStore, events...)
|
eventsToStore = append(eventsToStore, events...)
|
||||||
}
|
|
||||||
|
|
||||||
updateAccountPeers, err = areGroupChangesAffectPeers(ctx, transaction, accountID, groupIDs)
|
groupIDs = append(groupIDs, newGroup.ID)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
log.WithContext(ctx).Errorf("failed to update group %s: %v", newGroup.ID, err)
|
||||||
|
if len(groups) == 1 {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
globalErr = errors.Join(globalErr, err)
|
||||||
|
// continue updating other groups
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if err = transaction.UpdateGroups(ctx, accountID, groupsToSave); err != nil {
|
updateAccountPeers, err = areGroupChangesAffectPeers(ctx, am.Store, accountID, groupIDs)
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
return transaction.IncrementNetworkSerial(ctx, accountID)
|
|
||||||
})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -301,7 +321,7 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us
|
|||||||
am.UpdateAccountPeers(ctx, accountID)
|
am.UpdateAccountPeers(ctx, accountID)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return globalErr
|
||||||
}
|
}
|
||||||
|
|
||||||
// prepareGroupEvents prepares a list of event functions to be stored.
|
// prepareGroupEvents prepares a list of event functions to be stored.
|
||||||
@@ -584,13 +604,6 @@ func validateNewGroup(ctx context.Context, transaction store.Store, accountID st
|
|||||||
newGroup.ID = xid.New().String()
|
newGroup.ID = xid.New().String()
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, peerID := range newGroup.Peers {
|
|
||||||
_, err := transaction.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
|
|
||||||
if err != nil {
|
|
||||||
return status.Errorf(status.InvalidArgument, "peer with ID \"%s\" not found", peerID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+28
-15
@@ -519,33 +519,46 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
|
|||||||
initiatorUser = result
|
initiatorUser = result
|
||||||
}
|
}
|
||||||
|
|
||||||
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
|
var globalErr error
|
||||||
for _, update := range updates {
|
for _, update := range updates {
|
||||||
if update == nil {
|
if update == nil {
|
||||||
return status.Errorf(status.InvalidArgument, "provided user update is nil")
|
return nil, status.Errorf(status.InvalidArgument, "provided user update is nil")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
|
||||||
userHadPeers, updatedUser, userPeersToExpire, userEvents, err := am.processUserUpdate(
|
userHadPeers, updatedUser, userPeersToExpire, userEvents, err := am.processUserUpdate(
|
||||||
ctx, transaction, groupsMap, accountID, initiatorUserID, initiatorUser, update, addIfNotExists, settings,
|
ctx, transaction, groupsMap, accountID, initiatorUserID, initiatorUser, update, addIfNotExists, settings,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to process update for user %s: %w", update.Id, err)
|
return fmt.Errorf("failed to process update for user %s: %w", update.Id, err)
|
||||||
}
|
}
|
||||||
usersToSave = append(usersToSave, updatedUser)
|
|
||||||
addUserEvents = append(addUserEvents, userEvents...)
|
|
||||||
peersToExpire = append(peersToExpire, userPeersToExpire...)
|
|
||||||
|
|
||||||
if userHadPeers {
|
if userHadPeers {
|
||||||
updateAccountPeers = true
|
updateAccountPeers = true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
err = transaction.SaveUser(ctx, updatedUser)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to save updated user %s: %w", update.Id, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
usersToSave = append(usersToSave, updatedUser)
|
||||||
|
addUserEvents = append(addUserEvents, userEvents...)
|
||||||
|
peersToExpire = append(peersToExpire, userPeersToExpire...)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
log.WithContext(ctx).Errorf("failed to save user %s: %s", update.Id, err)
|
||||||
|
if len(updates) == 1 {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
globalErr = errors.Join(globalErr, err)
|
||||||
|
// continue when updating multiple users
|
||||||
}
|
}
|
||||||
return transaction.SaveUsers(ctx, usersToSave)
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
var updatedUsersInfo = make([]*types.UserInfo, 0, len(updates))
|
var updatedUsersInfo = make([]*types.UserInfo, 0, len(usersToSave))
|
||||||
|
|
||||||
userInfos, err := am.GetUsersFromAccount(ctx, accountID, initiatorUserID)
|
userInfos, err := am.GetUsersFromAccount(ctx, accountID, initiatorUserID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -578,7 +591,7 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
|
|||||||
am.UpdateAccountPeers(ctx, accountID)
|
am.UpdateAccountPeers(ctx, accountID)
|
||||||
}
|
}
|
||||||
|
|
||||||
return updatedUsersInfo, nil
|
return updatedUsersInfo, globalErr
|
||||||
}
|
}
|
||||||
|
|
||||||
// prepareUserUpdateEvents prepares a list user update events based on the changes between the old and new user data.
|
// prepareUserUpdateEvents prepares a list user update events based on the changes between the old and new user data.
|
||||||
@@ -643,7 +656,7 @@ func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transact
|
|||||||
}
|
}
|
||||||
transferredOwnerRole = result
|
transferredOwnerRole = result
|
||||||
|
|
||||||
userPeers, err := transaction.GetUserPeers(ctx, store.LockingStrengthUpdate, updatedUser.AccountID, update.Id)
|
userPeers, err := transaction.GetUserPeers(ctx, store.LockingStrengthNone, updatedUser.AccountID, update.Id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, nil, nil, nil, err
|
return false, nil, nil, nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user