mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-08-29 22:47:14 +00:00
fix(service): serialize client/inbound writes to prevent Postgres deadlock
Client/inbound mutations opened their own transactions that locked client_traffics before inbounds, while the @every 5s traffic poll (AddTraffic, already serialized through the traffic writer) locks them in the opposite order. Concurrently these formed an ABBA lock cycle that Postgres aborted as "deadlock detected" (SQLSTATE 40P01), failing client updates. Route those DB writes through the same single-goroutine traffic writer via a new runSerializedTx helper, so they can never run concurrently with the poll. For the client-edit paths the runtime (node) push is moved after the commit, keeping network I/O out of the serialized section. UpdateInbound keeps its push inside the transaction because EnsureInboundTagAllowed must reach the node before the central row is committed. Covers UpdateInboundClient/addInboundClient/DelInboundClientByEmail/ delInboundClients, the bulk adjust/delete transactions, and UpdateInbound.
This commit is contained in:
MHSanaei
+185
-178
@@ -966,195 +966,202 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
oldBits := inboundTransports(oldInbound.Protocol, oldInbound.StreamSettings, oldInbound.Settings)
|
||||
oldTagWasAuto := isAutoGeneratedTag(tag, oldInbound.Port, oldInbound.NodeID, oldBits)
|
||||
|
||||
db := database.GetDB()
|
||||
tx := db.Begin()
|
||||
|
||||
markDirty := false
|
||||
defer func() {
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
return
|
||||
}
|
||||
tx.Commit()
|
||||
if markDirty && oldInbound.NodeID != nil {
|
||||
if dErr := (&NodeService{}).MarkNodeDirty(*oldInbound.NodeID); dErr != nil {
|
||||
logger.Warning("mark node dirty failed:", dErr)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
err = s.updateClientTraffics(tx, oldInbound, inbound)
|
||||
if err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
|
||||
// Ensure created_at and updated_at exist in inbound.Settings clients
|
||||
{
|
||||
var oldSettings map[string]any
|
||||
_ = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
|
||||
emailToCreated := map[string]int64{}
|
||||
emailToUpdated := map[string]int64{}
|
||||
if oldSettings != nil {
|
||||
if oc, ok := oldSettings["clients"].([]any); ok {
|
||||
for _, it := range oc {
|
||||
if m, ok2 := it.(map[string]any); ok2 {
|
||||
if email, ok3 := m["email"].(string); ok3 {
|
||||
switch v := m["created_at"].(type) {
|
||||
case float64:
|
||||
emailToCreated[email] = int64(v)
|
||||
case int64:
|
||||
emailToCreated[email] = v
|
||||
}
|
||||
switch v := m["updated_at"].(type) {
|
||||
case float64:
|
||||
emailToUpdated[email] = int64(v)
|
||||
case int64:
|
||||
emailToUpdated[email] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
var newSettings map[string]any
|
||||
if err2 := json.Unmarshal([]byte(inbound.Settings), &newSettings); err2 == nil && newSettings != nil {
|
||||
now := time.Now().Unix() * 1000
|
||||
if nSlice, ok := newSettings["clients"].([]any); ok {
|
||||
for i := range nSlice {
|
||||
if m, ok2 := nSlice[i].(map[string]any); ok2 {
|
||||
email, _ := m["email"].(string)
|
||||
if _, ok3 := m["created_at"]; !ok3 {
|
||||
if v, ok4 := emailToCreated[email]; ok4 && v > 0 {
|
||||
m["created_at"] = v
|
||||
} else {
|
||||
m["created_at"] = now
|
||||
}
|
||||
}
|
||||
// Preserve client's updated_at if present; do not bump on parent inbound update
|
||||
if _, hasUpdated := m["updated_at"]; !hasUpdated {
|
||||
if v, ok4 := emailToUpdated[email]; ok4 && v > 0 {
|
||||
m["updated_at"] = v
|
||||
}
|
||||
}
|
||||
nSlice[i] = m
|
||||
}
|
||||
}
|
||||
newSettings["clients"] = nSlice
|
||||
if bs, err3 := json.MarshalIndent(newSettings, "", " "); err3 == nil {
|
||||
inbound.Settings = string(bs)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A Shadowsocks-2022 method change resizes the key, but existing client PSKs
|
||||
// keep their old length and would be rejected by xray. Regenerate mismatched
|
||||
// client keys so the inbound stays connectable.
|
||||
if normalized, changed := normalizeShadowsocksClientKeys(inbound.Settings); changed {
|
||||
inbound.Settings = normalized
|
||||
logger.Warning("Shadowsocks inbound", inbound.Id, "method change resized keys; regenerated mismatched client PSK(s)")
|
||||
}
|
||||
|
||||
oldInbound.Total = inbound.Total
|
||||
oldInbound.Remark = inbound.Remark
|
||||
oldInbound.SubSortIndex = inbound.SubSortIndex
|
||||
oldInbound.Enable = inbound.Enable
|
||||
oldInbound.ExpiryTime = inbound.ExpiryTime
|
||||
oldInbound.TrafficReset = inbound.TrafficReset
|
||||
oldInbound.Listen = inbound.Listen
|
||||
oldInbound.Port = inbound.Port
|
||||
oldInbound.Protocol = inbound.Protocol
|
||||
oldInbound.Settings = inbound.Settings
|
||||
oldInbound.StreamSettings = inbound.StreamSettings
|
||||
oldInbound.Sniffing = inbound.Sniffing
|
||||
if strings.TrimSpace(inbound.ShareAddrStrategy) == "" {
|
||||
normalizeInboundShareAddress(oldInbound)
|
||||
inbound.ShareAddrStrategy = oldInbound.ShareAddrStrategy
|
||||
inbound.ShareAddr = oldInbound.ShareAddr
|
||||
} else {
|
||||
if err := normalizeInboundShareAddressStrict(inbound); err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
oldInbound.ShareAddrStrategy = inbound.ShareAddrStrategy
|
||||
oldInbound.ShareAddr = inbound.ShareAddr
|
||||
}
|
||||
if oldTagWasAuto && inbound.Tag == tag {
|
||||
inbound.Tag = ""
|
||||
}
|
||||
oldInbound.Tag, err = s.resolveInboundTag(inbound, inbound.Id)
|
||||
if err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
inbound.Tag = oldInbound.Tag
|
||||
|
||||
needRestart := false
|
||||
rt, push, dirty, perr := s.nodePushPlan(oldInbound)
|
||||
if perr != nil {
|
||||
err = perr
|
||||
return inbound, false, err
|
||||
}
|
||||
if dirty {
|
||||
markDirty = true
|
||||
}
|
||||
if oldInbound.NodeID == nil {
|
||||
if !push {
|
||||
needRestart = true
|
||||
} else {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil {
|
||||
logger.Debug("Old inbound deleted on", rt.Name(), ":", tag)
|
||||
markDirty := false
|
||||
|
||||
// Persist the client-stat sync, settings munging, runtime push and inbound
|
||||
// save as one transaction routed through the serial traffic writer, so it
|
||||
// never runs concurrently with the @every 5s traffic poll. Both touch
|
||||
// client_traffics and inbounds in opposite order, which Postgres aborts as a
|
||||
// deadlock (40P01); serializing removes the contention (runSerializedTx).
|
||||
//
|
||||
// The runtime push stays inside the transaction here (unlike the client-edit
|
||||
// paths that apply it after commit): EnsureInboundTagAllowed must reach the
|
||||
// node before the central row is committed, or a "selected"-mode node would
|
||||
// sweep the renamed inbound on its next pull. Inbound edits are rare, so
|
||||
// holding the writer across the node call is an acceptable trade.
|
||||
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
||||
if err := s.updateClientTraffics(tx, oldInbound, inbound); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Ensure created_at and updated_at exist in inbound.Settings clients
|
||||
{
|
||||
var oldSettings map[string]any
|
||||
_ = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
|
||||
emailToCreated := map[string]int64{}
|
||||
emailToUpdated := map[string]int64{}
|
||||
if oldSettings != nil {
|
||||
if oc, ok := oldSettings["clients"].([]any); ok {
|
||||
for _, it := range oc {
|
||||
if m, ok2 := it.(map[string]any); ok2 {
|
||||
if email, ok3 := m["email"].(string); ok3 {
|
||||
switch v := m["created_at"].(type) {
|
||||
case float64:
|
||||
emailToCreated[email] = int64(v)
|
||||
case int64:
|
||||
emailToCreated[email] = v
|
||||
}
|
||||
switch v := m["updated_at"].(type) {
|
||||
case float64:
|
||||
emailToUpdated[email] = int64(v)
|
||||
case int64:
|
||||
emailToUpdated[email] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if inbound.Enable {
|
||||
runtimeInbound, err2 := s.buildRuntimeInboundForAPI(tx, oldInbound)
|
||||
if err2 != nil {
|
||||
logger.Debug("Unable to prepare runtime inbound config:", err2)
|
||||
needRestart = true
|
||||
} else if err2 := rt.AddInbound(context.Background(), runtimeInbound); err2 == nil {
|
||||
logger.Debug("Updated inbound added on", rt.Name(), ":", oldInbound.Tag)
|
||||
} else {
|
||||
logger.Debug("Unable to update inbound on", rt.Name(), ":", err2)
|
||||
needRestart = true
|
||||
var newSettings map[string]any
|
||||
if err2 := json.Unmarshal([]byte(inbound.Settings), &newSettings); err2 == nil && newSettings != nil {
|
||||
now := time.Now().Unix() * 1000
|
||||
if nSlice, ok := newSettings["clients"].([]any); ok {
|
||||
for i := range nSlice {
|
||||
if m, ok2 := nSlice[i].(map[string]any); ok2 {
|
||||
email, _ := m["email"].(string)
|
||||
if _, ok3 := m["created_at"]; !ok3 {
|
||||
if v, ok4 := emailToCreated[email]; ok4 && v > 0 {
|
||||
m["created_at"] = v
|
||||
} else {
|
||||
m["created_at"] = now
|
||||
}
|
||||
}
|
||||
// Preserve client's updated_at if present; do not bump on parent inbound update
|
||||
if _, hasUpdated := m["updated_at"]; !hasUpdated {
|
||||
if v, ok4 := emailToUpdated[email]; ok4 && v > 0 {
|
||||
m["updated_at"] = v
|
||||
}
|
||||
}
|
||||
nSlice[i] = m
|
||||
}
|
||||
}
|
||||
newSettings["clients"] = nSlice
|
||||
if bs, err3 := json.MarshalIndent(newSettings, "", " "); err3 == nil {
|
||||
inbound.Settings = string(bs)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if push {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
if !inbound.Enable {
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 != nil {
|
||||
logger.Warning("Unable to disable inbound on", rt.Name(), ":", err2)
|
||||
markDirty = true
|
||||
|
||||
// A Shadowsocks-2022 method change resizes the key, but existing client PSKs
|
||||
// keep their old length and would be rejected by xray. Regenerate mismatched
|
||||
// client keys so the inbound stays connectable.
|
||||
if normalized, changed := normalizeShadowsocksClientKeys(inbound.Settings); changed {
|
||||
inbound.Settings = normalized
|
||||
logger.Warning("Shadowsocks inbound", inbound.Id, "method change resized keys; regenerated mismatched client PSK(s)")
|
||||
}
|
||||
|
||||
oldInbound.Total = inbound.Total
|
||||
oldInbound.Remark = inbound.Remark
|
||||
oldInbound.SubSortIndex = inbound.SubSortIndex
|
||||
oldInbound.Enable = inbound.Enable
|
||||
oldInbound.ExpiryTime = inbound.ExpiryTime
|
||||
oldInbound.TrafficReset = inbound.TrafficReset
|
||||
oldInbound.Listen = inbound.Listen
|
||||
oldInbound.Port = inbound.Port
|
||||
oldInbound.Protocol = inbound.Protocol
|
||||
oldInbound.Settings = inbound.Settings
|
||||
oldInbound.StreamSettings = inbound.StreamSettings
|
||||
oldInbound.Sniffing = inbound.Sniffing
|
||||
if strings.TrimSpace(inbound.ShareAddrStrategy) == "" {
|
||||
normalizeInboundShareAddress(oldInbound)
|
||||
inbound.ShareAddrStrategy = oldInbound.ShareAddrStrategy
|
||||
inbound.ShareAddr = oldInbound.ShareAddr
|
||||
} else {
|
||||
if err := normalizeInboundShareAddressStrict(inbound); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, oldInbound); err2 != nil {
|
||||
logger.Warning("Unable to update inbound on", rt.Name(), ":", err2)
|
||||
oldInbound.ShareAddrStrategy = inbound.ShareAddrStrategy
|
||||
oldInbound.ShareAddr = inbound.ShareAddr
|
||||
}
|
||||
if oldTagWasAuto && inbound.Tag == tag {
|
||||
inbound.Tag = ""
|
||||
}
|
||||
resolvedTag, err := s.resolveInboundTag(inbound, inbound.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
oldInbound.Tag = resolvedTag
|
||||
inbound.Tag = oldInbound.Tag
|
||||
|
||||
rt, push, dirty, perr := s.nodePushPlan(oldInbound)
|
||||
if perr != nil {
|
||||
return perr
|
||||
}
|
||||
if dirty {
|
||||
markDirty = true
|
||||
}
|
||||
}
|
||||
|
||||
// A rename must allow the new tag before the deferred commit, or a node in
|
||||
// "selected" sync mode would sweep the renamed central row on the next pull.
|
||||
if oldInbound.NodeID != nil {
|
||||
if aErr := (&NodeService{}).EnsureInboundTagAllowed(*oldInbound.NodeID, oldInbound.Tag); aErr != nil {
|
||||
logger.Warning("allow inbound tag on node failed:", aErr)
|
||||
if oldInbound.NodeID == nil {
|
||||
if !push {
|
||||
needRestart = true
|
||||
} else {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil {
|
||||
logger.Debug("Old inbound deleted on", rt.Name(), ":", tag)
|
||||
}
|
||||
if inbound.Enable {
|
||||
runtimeInbound, err2 := s.buildRuntimeInboundForAPI(tx, oldInbound)
|
||||
if err2 != nil {
|
||||
logger.Debug("Unable to prepare runtime inbound config:", err2)
|
||||
needRestart = true
|
||||
} else if err2 := rt.AddInbound(context.Background(), runtimeInbound); err2 == nil {
|
||||
logger.Debug("Updated inbound added on", rt.Name(), ":", oldInbound.Tag)
|
||||
} else {
|
||||
logger.Debug("Unable to update inbound on", rt.Name(), ":", err2)
|
||||
needRestart = true
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if push {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
if !inbound.Enable {
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 != nil {
|
||||
logger.Warning("Unable to disable inbound on", rt.Name(), ":", err2)
|
||||
markDirty = true
|
||||
}
|
||||
} else if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, oldInbound); err2 != nil {
|
||||
logger.Warning("Unable to update inbound on", rt.Name(), ":", err2)
|
||||
markDirty = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err = tx.Save(oldInbound).Error; err != nil {
|
||||
return inbound, false, err
|
||||
// A rename must allow the new tag before the inbound row is committed, or a
|
||||
// node in "selected" sync mode would sweep the renamed central row on the
|
||||
// next pull.
|
||||
if oldInbound.NodeID != nil {
|
||||
if aErr := (&NodeService{}).EnsureInboundTagAllowed(*oldInbound.NodeID, oldInbound.Tag); aErr != nil {
|
||||
logger.Warning("allow inbound tag on node failed:", aErr)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Save(oldInbound).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
newClients, gcErr := s.GetClients(oldInbound)
|
||||
if gcErr != nil {
|
||||
return gcErr
|
||||
}
|
||||
if err := s.clientService.SyncInbound(tx, oldInbound.Id, newClients); err != nil {
|
||||
return err
|
||||
}
|
||||
// (Re)generate the Xray config whenever routing was or is now enabled, so
|
||||
// the egress SOCKS bridge is added, moved, or dropped to match the new
|
||||
// settings.
|
||||
if mtprotoRoutesThroughXray(inbound) || oldRoutedMtproto {
|
||||
needRestart = true
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if txErr != nil {
|
||||
return inbound, false, txErr
|
||||
}
|
||||
newClients, gcErr := s.GetClients(oldInbound)
|
||||
if gcErr != nil {
|
||||
err = gcErr
|
||||
return inbound, false, err
|
||||
}
|
||||
if err = s.clientService.SyncInbound(tx, oldInbound.Id, newClients); err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
// (Re)generate the Xray config whenever routing was or is now enabled, so the
|
||||
// egress SOCKS bridge is added, moved, or dropped to match the new settings.
|
||||
if mtprotoRoutesThroughXray(inbound) || oldRoutedMtproto {
|
||||
needRestart = true
|
||||
if markDirty && oldInbound.NodeID != nil {
|
||||
if dErr := (&NodeService{}).MarkNodeDirty(*oldInbound.NodeID); dErr != nil {
|
||||
logger.Warning("mark node dirty failed:", dErr)
|
||||
}
|
||||
}
|
||||
return inbound, needRestart, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user