|
@@ -107,7 +107,6 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
needRestart := false
|
|
needRestart := false
|
|
|
- markDirty := false
|
|
|
|
|
|
|
|
|
|
// Read each client's live state before the DB write (DelClientStat would
|
|
// Read each client's live state before the DB write (DelClientStat would
|
|
|
// erase the enable flag we need to decide on a runtime removal).
|
|
// erase the enable flag we need to decide on a runtime removal).
|
|
@@ -158,7 +157,13 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|
|
if gcErr != nil {
|
|
if gcErr != nil {
|
|
|
return gcErr
|
|
return gcErr
|
|
|
}
|
|
}
|
|
|
- return s.SyncInbound(tx, inboundId, finalClients)
|
|
|
|
|
|
|
+ if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
|
|
|
|
|
+ return err
|
|
|
|
|
+ }
|
|
|
|
|
+ if oldInbound.NodeID != nil {
|
|
|
|
|
+ return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
}); txErr != nil {
|
|
}); txErr != nil {
|
|
|
return needRestart, txErr
|
|
return needRestart, txErr
|
|
|
}
|
|
}
|
|
@@ -167,17 +172,13 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|
|
var nodeRt runtime.Runtime
|
|
var nodeRt runtime.Runtime
|
|
|
nodePush := false
|
|
nodePush := false
|
|
|
if oldInbound.NodeID != nil {
|
|
if oldInbound.NodeID != nil {
|
|
|
- rt, push, dirty, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
|
|
|
|
|
+ rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
|
if perr != nil {
|
|
if perr != nil {
|
|
|
return needRestart, perr
|
|
return needRestart, perr
|
|
|
}
|
|
}
|
|
|
- if dirty {
|
|
|
|
|
- markDirty = true
|
|
|
|
|
- }
|
|
|
|
|
nodeRt, nodePush = rt, push
|
|
nodeRt, nodePush = rt, push
|
|
|
// Large batches collapse into one reconcile push rather than M deletes.
|
|
// Large batches collapse into one reconcile push rather than M deletes.
|
|
|
if nodePush && len(targets) > nodeBulkPushThreshold {
|
|
if nodePush && len(targets) > nodeBulkPushThreshold {
|
|
|
- markDirty = true
|
|
|
|
|
nodePush = false
|
|
nodePush = false
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
@@ -202,16 +203,10 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|
|
} else if nodePush {
|
|
} else if nodePush {
|
|
|
if err1 := nodeRt.DeleteUser(context.Background(), oldInbound, t.email); err1 != nil {
|
|
if err1 := nodeRt.DeleteUser(context.Background(), oldInbound, t.email); err1 != nil {
|
|
|
logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1)
|
|
logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1)
|
|
|
- markDirty = true
|
|
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- if markDirty && oldInbound.NodeID != nil {
|
|
|
|
|
- if dErr := (&NodeService{}).MarkNodeDirty(*oldInbound.NodeID); dErr != nil {
|
|
|
|
|
- logger.Warning("mark node dirty failed:", dErr)
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
return needRestart, nil
|
|
return needRestart, nil
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -357,15 +352,11 @@ func (s *ClientService) addInboundClient(inboundSvc *InboundService, data *model
|
|
|
oldInbound.Settings = string(newSettings)
|
|
oldInbound.Settings = string(newSettings)
|
|
|
|
|
|
|
|
needRestart := false
|
|
needRestart := false
|
|
|
- markDirty := false
|
|
|
|
|
|
|
|
|
|
- rt, push, dirty, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
|
|
|
|
|
+ rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
|
if perr != nil {
|
|
if perr != nil {
|
|
|
return false, perr
|
|
return false, perr
|
|
|
}
|
|
}
|
|
|
- if dirty {
|
|
|
|
|
- markDirty = true
|
|
|
|
|
- }
|
|
|
|
|
|
|
|
|
|
// Persist client stats + inbound atomically, serialized against the traffic
|
|
// Persist client stats + inbound atomically, serialized against the traffic
|
|
|
// poll to avoid the cross-transaction lock-order deadlock (runSerializedTx).
|
|
// poll to avoid the cross-transaction lock-order deadlock (runSerializedTx).
|
|
@@ -385,7 +376,13 @@ func (s *ClientService) addInboundClient(inboundSvc *InboundService, data *model
|
|
|
if gcErr != nil {
|
|
if gcErr != nil {
|
|
|
return gcErr
|
|
return gcErr
|
|
|
}
|
|
}
|
|
|
- return s.SyncInbound(tx, oldInbound.Id, finalClients)
|
|
|
|
|
|
|
+ if err := s.SyncInbound(tx, oldInbound.Id, finalClients); err != nil {
|
|
|
|
|
+ return err
|
|
|
|
|
+ }
|
|
|
|
|
+ if oldInbound.NodeID != nil {
|
|
|
|
|
+ return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
}); txErr != nil {
|
|
}); txErr != nil {
|
|
|
return false, txErr
|
|
return false, txErr
|
|
|
}
|
|
}
|
|
@@ -434,25 +431,18 @@ func (s *ClientService) addInboundClient(inboundSvc *InboundService, data *model
|
|
|
// settings already hold the final set, so mark dirty and let one reconcile
|
|
// settings already hold the final set, so mark dirty and let one reconcile
|
|
|
// push converge the node instead.
|
|
// push converge the node instead.
|
|
|
if push && len(clients) > nodeBulkPushThreshold {
|
|
if push && len(clients) > nodeBulkPushThreshold {
|
|
|
- markDirty = true
|
|
|
|
|
push = false
|
|
push = false
|
|
|
}
|
|
}
|
|
|
for _, client := range clients {
|
|
for _, client := range clients {
|
|
|
if push {
|
|
if push {
|
|
|
if err1 := rt.AddClient(context.Background(), oldInbound, client); err1 != nil {
|
|
if err1 := rt.AddClient(context.Background(), oldInbound, client); err1 != nil {
|
|
|
logger.Warning("Error in adding client on", rt.Name(), ":", err1)
|
|
logger.Warning("Error in adding client on", rt.Name(), ":", err1)
|
|
|
- markDirty = true
|
|
|
|
|
push = false
|
|
push = false
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- if markDirty && oldInbound.NodeID != nil {
|
|
|
|
|
- if dErr := (&NodeService{}).MarkNodeDirty(*oldInbound.NodeID); dErr != nil {
|
|
|
|
|
- logger.Warning("mark node dirty failed:", dErr)
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
return needRestart, nil
|
|
return needRestart, nil
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -623,7 +613,6 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|
|
oldInbound.Settings = string(newSettings)
|
|
oldInbound.Settings = string(newSettings)
|
|
|
|
|
|
|
|
needRestart := false
|
|
needRestart := false
|
|
|
- markDirty := false
|
|
|
|
|
|
|
|
|
|
// Resolve the push plan before the DB write so a node-state lookup failure
|
|
// Resolve the push plan before the DB write so a node-state lookup failure
|
|
|
// still aborts the whole update without committing anything (it used to roll
|
|
// still aborts the whole update without committing anything (it used to roll
|
|
@@ -631,15 +620,11 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|
|
var rt runtime.Runtime
|
|
var rt runtime.Runtime
|
|
|
var push bool
|
|
var push bool
|
|
|
if len(oldEmail) > 0 {
|
|
if len(oldEmail) > 0 {
|
|
|
- var dirty bool
|
|
|
|
|
var perr error
|
|
var perr error
|
|
|
- rt, push, dirty, perr = inboundSvc.nodePushPlan(oldInbound)
|
|
|
|
|
|
|
+ rt, push, _, perr = inboundSvc.nodePushPlan(oldInbound)
|
|
|
if perr != nil {
|
|
if perr != nil {
|
|
|
return false, perr
|
|
return false, perr
|
|
|
}
|
|
}
|
|
|
- if dirty {
|
|
|
|
|
- markDirty = true
|
|
|
|
|
- }
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// Persist client stats + inbound atomically, serialized against the traffic
|
|
// Persist client stats + inbound atomically, serialized against the traffic
|
|
@@ -705,7 +690,13 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|
|
if gcErr != nil {
|
|
if gcErr != nil {
|
|
|
return gcErr
|
|
return gcErr
|
|
|
}
|
|
}
|
|
|
- return s.SyncInbound(tx, oldInbound.Id, finalClients)
|
|
|
|
|
|
|
+ if err := s.SyncInbound(tx, oldInbound.Id, finalClients); err != nil {
|
|
|
|
|
+ return err
|
|
|
|
|
+ }
|
|
|
|
|
+ if oldInbound.NodeID != nil {
|
|
|
|
|
+ return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
}); txErr != nil {
|
|
}); txErr != nil {
|
|
|
return false, txErr
|
|
return false, txErr
|
|
|
}
|
|
}
|
|
@@ -757,7 +748,6 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|
|
} else if push {
|
|
} else if push {
|
|
|
if err1 := rt.UpdateUser(context.Background(), oldInbound, oldEmail, clients[0]); err1 != nil {
|
|
if err1 := rt.UpdateUser(context.Background(), oldInbound, oldEmail, clients[0]); err1 != nil {
|
|
|
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
|
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
|
|
- markDirty = true
|
|
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
} else {
|
|
} else {
|
|
@@ -765,11 +755,6 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|
|
needRestart = true
|
|
needRestart = true
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- if markDirty && oldInbound.NodeID != nil {
|
|
|
|
|
- if dErr := (&NodeService{}).MarkNodeDirty(*oldInbound.NodeID); dErr != nil {
|
|
|
|
|
- logger.Warning("mark node dirty failed:", dErr)
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
return needRestart, nil
|
|
return needRestart, nil
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -831,7 +816,6 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
needRestart := false
|
|
needRestart := false
|
|
|
- markDirty := false
|
|
|
|
|
|
|
|
|
|
// Decide what to delete and the push plan before the serialized DB write —
|
|
// Decide what to delete and the push plan before the serialized DB write —
|
|
|
// these are reads, and nodePushPlan failing should abort before committing.
|
|
// these are reads, and nodePushPlan failing should abort before committing.
|
|
@@ -850,14 +834,11 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|
|
var rt runtime.Runtime
|
|
var rt runtime.Runtime
|
|
|
var push bool
|
|
var push bool
|
|
|
if len(email) > 0 && (oldInbound.NodeID != nil || needApiDel) {
|
|
if len(email) > 0 && (oldInbound.NodeID != nil || needApiDel) {
|
|
|
- r, p, dirty, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
|
|
|
|
|
+ r, p, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
|
if perr != nil {
|
|
if perr != nil {
|
|
|
return false, perr
|
|
return false, perr
|
|
|
}
|
|
}
|
|
|
rt, push = r, p
|
|
rt, push = r, p
|
|
|
- if dirty {
|
|
|
|
|
- markDirty = true
|
|
|
|
|
- }
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// Persist the deletion atomically, serialized against the traffic poll to
|
|
// Persist the deletion atomically, serialized against the traffic poll to
|
|
@@ -882,7 +863,13 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|
|
if gcErr != nil {
|
|
if gcErr != nil {
|
|
|
return gcErr
|
|
return gcErr
|
|
|
}
|
|
}
|
|
|
- return s.SyncInbound(tx, inboundId, finalClients)
|
|
|
|
|
|
|
+ if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
|
|
|
|
|
+ return err
|
|
|
|
|
+ }
|
|
|
|
|
+ if oldInbound.NodeID != nil {
|
|
|
|
|
+ return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
}); txErr != nil {
|
|
}); txErr != nil {
|
|
|
return false, txErr
|
|
return false, txErr
|
|
|
}
|
|
}
|
|
@@ -915,17 +902,11 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|
|
if push {
|
|
if push {
|
|
|
if err1 := rt.DeleteUser(context.Background(), oldInbound, email); err1 != nil {
|
|
if err1 := rt.DeleteUser(context.Background(), oldInbound, email); err1 != nil {
|
|
|
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
|
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
|
|
- markDirty = true
|
|
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- if markDirty && oldInbound.NodeID != nil {
|
|
|
|
|
- if dErr := (&NodeService{}).MarkNodeDirty(*oldInbound.NodeID); dErr != nil {
|
|
|
|
|
- logger.Warning("mark node dirty failed:", dErr)
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
return needRestart, nil
|
|
return needRestart, nil
|
|
|
}
|
|
}
|
|
|
|
|
|