mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-10-29 06:31:58 +00:00
fix(clients): stop client ops from reverting a renewal committed mid-op
Invariant: an operation on an inbound's clients writes back only what it changed, onto the settings as committed when it writes. Every client op (add, edit, delete, bulk adjust/detach/delete/set-enable) read the inbound outside the serial traffic writer and then tx.Save'd the whole row inside it. A traffic tick that committed in between - auto-renew re-enabling a client, the delayed-start conversion, a node adoption - was overwritten with the stale copy. A renewed neighbour ended up enable=false with its old expiry in settings while client_traffics said enabled, so the next runtime rebuild dropped a healthy client nobody had touched. The bulk ops then ran a full SyncInbound from those stale settings, copying enable=false into the client record too. Each site now commits through commitInboundClientSettings: inside the serialized tx it three-way merges the op's edit (read -> output, per client and per field) onto the committed settings and updates only the settings column, which also stops the stale up/down/enable inbound columns being written back. advancePushedInbound now records what the per-client push delivered rather than the merged settings, so the node's reconcile-skip fingerprint cannot claim a renewal it never received.
This commit is contained in:
1 parent
15d82a5e47
commit
63ffc083e4
4 files changed
+497
-16
No files matched your search
@@ -795,6 +795,7 @@ func (s *ClientService) bulkAdjustInboundClients(
|
|||||||
}
|
}
|
||||||
return res
|
return res
|
||||||
}
|
}
|
||||||
|
prevSettings := oldInbound.Settings
|
||||||
oldInbound.Settings = string(newSettings)
|
oldInbound.Settings = string(newSettings)
|
||||||
|
|
||||||
// A flow change rewrites the user's xray config, which the lightweight
|
// A flow change rewrites the user's xray config, which the lightweight
|
||||||
@@ -807,7 +808,7 @@ func (s *ClientService) bulkAdjustInboundClients(
|
|||||||
// Serialize against the traffic poll to avoid the cross-transaction
|
// Serialize against the traffic poll to avoid the cross-transaction
|
||||||
// lock-order deadlock on inbounds/client_records (runSerializedTx).
|
// lock-order deadlock on inbounds/client_records (runSerializedTx).
|
||||||
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
||||||
if err := tx.Save(oldInbound).Error; err != nil {
|
if err := commitInboundClientSettings(tx, oldInbound, prevSettings); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
finalClients, gcErr := inboundSvc.GetClients(oldInbound)
|
finalClients, gcErr := inboundSvc.GetClients(oldInbound)
|
||||||
@@ -1112,6 +1113,7 @@ func (s *ClientService) bulkDelInboundClients(
|
|||||||
}
|
}
|
||||||
return res
|
return res
|
||||||
}
|
}
|
||||||
|
prevSettings := oldInbound.Settings
|
||||||
oldInbound.Settings = string(newSettings)
|
oldInbound.Settings = string(newSettings)
|
||||||
|
|
||||||
foundList := make([]string, 0, len(foundEmails))
|
foundList := make([]string, 0, len(foundEmails))
|
||||||
@@ -1181,7 +1183,7 @@ func (s *ClientService) bulkDelInboundClients(
|
|||||||
// Serialize against the traffic poll to avoid the cross-transaction
|
// Serialize against the traffic poll to avoid the cross-transaction
|
||||||
// lock-order deadlock on inbounds/client_records (runSerializedTx).
|
// lock-order deadlock on inbounds/client_records (runSerializedTx).
|
||||||
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
||||||
if err := tx.Save(oldInbound).Error; err != nil {
|
if err := commitInboundClientSettings(tx, oldInbound, prevSettings); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
finalClients, err := inboundSvc.GetClients(oldInbound)
|
finalClients, err := inboundSvc.GetClients(oldInbound)
|
||||||
@@ -1780,7 +1782,7 @@ func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService,
|
|||||||
}
|
}
|
||||||
|
|
||||||
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
||||||
if e := tx.Save(oldInbound).Error; e != nil {
|
if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
finalClients, gcErr := inboundSvc.GetClients(oldInbound)
|
finalClients, gcErr := inboundSvc.GetClients(oldInbound)
|
||||||
@@ -1850,7 +1852,7 @@ func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if !pushFailed {
|
if !pushFailed {
|
||||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -39,16 +39,18 @@ func sameClientConfigExceptUpdatedAt(a, b map[string]any) bool {
|
|||||||
return aerr == nil && berr == nil && string(an) == string(bn)
|
return aerr == nil && berr == nil && string(an) == string(bn)
|
||||||
}
|
}
|
||||||
|
|
||||||
// advancePushedInbound advances the node's reconcile-skip fingerprint from the
|
// advancePushedInbound advances the node's reconcile-skip fingerprint to what the
|
||||||
// pre-edit settings to the saved ones after every per-client push succeeded.
|
// per-client pushes delivered, not the saved settings a traffic tick may have extended.
|
||||||
func advancePushedInbound(rt runtime.Runtime, prevSettings string, ib *model.Inbound) {
|
func advancePushedInbound(rt runtime.Runtime, prevSettings, pushedSettings string, ib *model.Inbound) {
|
||||||
rem, ok := rt.(*runtime.Remote)
|
rem, ok := rt.(*runtime.Remote)
|
||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
prev := *ib
|
prev := *ib
|
||||||
prev.Settings = prevSettings
|
prev.Settings = prevSettings
|
||||||
rem.AdvancePushedInbound(&prev, ib)
|
pushed := *ib
|
||||||
|
pushed.Settings = pushedSettings
|
||||||
|
rem.AdvancePushedInbound(&prev, &pushed)
|
||||||
}
|
}
|
||||||
|
|
||||||
// delInboundClients removes several clients from a single inbound in one pass:
|
// delInboundClients removes several clients from a single inbound in one pass:
|
||||||
@@ -184,7 +186,7 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if e := tx.Save(oldInbound).Error; e != nil {
|
if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
detached := make([]string, 0, len(targets))
|
detached := make([]string, 0, len(targets))
|
||||||
@@ -249,7 +251,7 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if nodePush && !nodePushFailed {
|
if nodePush && !nodePushFailed {
|
||||||
advancePushedInbound(nodeRt, prevSettings, oldInbound)
|
advancePushedInbound(nodeRt, prevSettings, string(newSettings), oldInbound)
|
||||||
}
|
}
|
||||||
|
|
||||||
return needRestart, nil
|
return needRestart, nil
|
||||||
@@ -563,7 +565,7 @@ func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model
|
|||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if e := tx.Save(oldInbound).Error; e != nil {
|
if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
if err := s.ApplyInboundClientDelta(tx, oldInbound.Id, addedClients, nil); err != nil {
|
if err := s.ApplyInboundClientDelta(tx, oldInbound.Id, addedClients, nil); err != nil {
|
||||||
@@ -648,7 +650,7 @@ func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if push {
|
if push {
|
||||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -980,7 +982,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if e := tx.Save(oldInbound).Error; e != nil {
|
if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
// Rename the client record in the same transaction as the settings JSON
|
// Rename the client record in the same transaction as the settings JSON
|
||||||
@@ -1074,7 +1076,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|||||||
if err1 != nil {
|
if err1 != nil {
|
||||||
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
||||||
} else {
|
} else {
|
||||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
@@ -1184,7 +1186,7 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if e := tx.Save(oldInbound).Error; e != nil {
|
if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
if err := s.ApplyInboundClientDelta(tx, inboundId, nil, []string{email}); err != nil {
|
if err := s.ApplyInboundClientDelta(tx, inboundId, nil, []string{email}); err != nil {
|
||||||
@@ -1248,7 +1250,7 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|||||||
if err1 != nil {
|
if err1 != nil {
|
||||||
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
||||||
} else {
|
} else {
|
||||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,141 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"reflect"
|
||||||
|
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||||
|
|
||||||
|
"gorm.io/gorm"
|
||||||
|
)
|
||||||
|
|
||||||
|
// commitInboundClientSettings writes a client op's edit (base → ib.Settings) onto
|
||||||
|
// the settings committed now: a traffic tick after the op's read must survive.
|
||||||
|
func commitInboundClientSettings(tx *gorm.DB, ib *model.Inbound, base string) error {
|
||||||
|
var current []string
|
||||||
|
if err := tx.Model(&model.Inbound{}).Where("id = ?", ib.Id).Pluck("settings", ¤t).Error; err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if len(current) == 1 && current[0] != base {
|
||||||
|
merged, err := rebaseClientSettings(base, ib.Settings, current[0])
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
ib.Settings = merged
|
||||||
|
}
|
||||||
|
return tx.Model(&model.Inbound{}).Where("id = ?", ib.Id).Update("settings", ib.Settings).Error
|
||||||
|
}
|
||||||
|
|
||||||
|
// rebaseClientSettings three-way merges settings JSON: every key and client field
|
||||||
|
// ours left as base had it takes current's value; clients are matched by email.
|
||||||
|
func rebaseClientSettings(base, ours, current string) (string, error) {
|
||||||
|
var baseM, oursM, curM map[string]any
|
||||||
|
for _, p := range []struct {
|
||||||
|
raw string
|
||||||
|
dst *map[string]any
|
||||||
|
}{{base, &baseM}, {ours, &oursM}, {current, &curM}} {
|
||||||
|
if err := json.Unmarshal([]byte(p.raw), p.dst); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out := mergeFields(baseM, oursM, curM)
|
||||||
|
baseClients, _ := baseM["clients"].([]any)
|
||||||
|
oursClients, _ := oursM["clients"].([]any)
|
||||||
|
curClients, _ := curM["clients"].([]any)
|
||||||
|
if _, has := oursM["clients"]; has {
|
||||||
|
out["clients"] = mergeClientLists(baseClients, oursClients, curClients)
|
||||||
|
}
|
||||||
|
b, err := json.MarshalIndent(out, "", " ")
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return string(b), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func mergeFields(base, ours, current map[string]any) map[string]any {
|
||||||
|
out := make(map[string]any, len(current)+len(ours))
|
||||||
|
for k, v := range current {
|
||||||
|
out[k] = v
|
||||||
|
}
|
||||||
|
keys := make(map[string]struct{}, len(base)+len(ours))
|
||||||
|
for k := range base {
|
||||||
|
keys[k] = struct{}{}
|
||||||
|
}
|
||||||
|
for k := range ours {
|
||||||
|
keys[k] = struct{}{}
|
||||||
|
}
|
||||||
|
for k := range keys {
|
||||||
|
bv, inBase := base[k]
|
||||||
|
ov, inOurs := ours[k]
|
||||||
|
if inBase == inOurs && reflect.DeepEqual(bv, ov) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if inOurs {
|
||||||
|
out[k] = ov
|
||||||
|
} else {
|
||||||
|
delete(out, k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
func clientEntryEmail(entry any) (map[string]any, string) {
|
||||||
|
m, ok := entry.(map[string]any)
|
||||||
|
if !ok {
|
||||||
|
return nil, ""
|
||||||
|
}
|
||||||
|
email, _ := m["email"].(string)
|
||||||
|
return m, email
|
||||||
|
}
|
||||||
|
|
||||||
|
func indexClientsByEmail(list []any) map[string]map[string]any {
|
||||||
|
out := make(map[string]map[string]any, len(list))
|
||||||
|
for _, entry := range list {
|
||||||
|
if m, email := clientEntryEmail(entry); email != "" {
|
||||||
|
out[email] = m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// mergeClientLists keeps ours' order. A client ours removed stays removed; one a
|
||||||
|
// concurrent writer added or removed keeps that change unless ours edited it.
|
||||||
|
func mergeClientLists(base, ours, current []any) []any {
|
||||||
|
baseBy := indexClientsByEmail(base)
|
||||||
|
curBy := indexClientsByEmail(current)
|
||||||
|
out := make([]any, 0, len(ours)+len(current))
|
||||||
|
placed := make(map[string]struct{}, len(ours))
|
||||||
|
for _, entry := range ours {
|
||||||
|
o, email := clientEntryEmail(entry)
|
||||||
|
if email == "" {
|
||||||
|
out = append(out, entry)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
placed[email] = struct{}{}
|
||||||
|
b, inBase := baseBy[email]
|
||||||
|
c, inCur := curBy[email]
|
||||||
|
switch {
|
||||||
|
case !inBase:
|
||||||
|
out = append(out, o)
|
||||||
|
case !inCur:
|
||||||
|
if !reflect.DeepEqual(b, o) {
|
||||||
|
out = append(out, o)
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
out = append(out, mergeFields(b, o, c))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, entry := range current {
|
||||||
|
c, email := clientEntryEmail(entry)
|
||||||
|
if email == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, done := placed[email]; done {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, inBase := baseBy[email]; !inBase {
|
||||||
|
out = append(out, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
@@ -0,0 +1,336 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"net/url"
|
||||||
|
"reflect"
|
||||||
|
"strconv"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
||||||
|
|
||||||
|
"gorm.io/gorm"
|
||||||
|
)
|
||||||
|
|
||||||
|
// commitTickBetweenReadAndWrite parks the serial writer, lets op read the
|
||||||
|
// inbound and queue its transaction, then commits tick ahead of that transaction.
|
||||||
|
func commitTickBetweenReadAndWrite(t *testing.T, tick func(tx *gorm.DB) error, op func()) {
|
||||||
|
t.Helper()
|
||||||
|
resetTrafficWriterForTest(t)
|
||||||
|
StartTrafficWriter()
|
||||||
|
|
||||||
|
parked := make(chan struct{})
|
||||||
|
release := make(chan struct{})
|
||||||
|
tickErr := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
tickErr <- submitTrafficWrite(func() error {
|
||||||
|
close(parked)
|
||||||
|
<-release
|
||||||
|
return database.GetDB().Transaction(tick)
|
||||||
|
})
|
||||||
|
}()
|
||||||
|
<-parked
|
||||||
|
|
||||||
|
opDone := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
defer close(opDone)
|
||||||
|
op()
|
||||||
|
}()
|
||||||
|
waitTrafficWriterQueued(t)
|
||||||
|
close(release)
|
||||||
|
if err := <-tickErr; err != nil {
|
||||||
|
t.Fatalf("tick: %v", err)
|
||||||
|
}
|
||||||
|
<-opDone
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedRenewableNeighbour builds an inbound holding a healthy client X and a
|
||||||
|
// quota-disabled client Y whose auto-renew is due, as the traffic job sees them.
|
||||||
|
func seedRenewableNeighbour(t *testing.T, port int, nodeID *int) *model.Inbound {
|
||||||
|
t.Helper()
|
||||||
|
past := time.Now().Add(-time.Hour).UnixMilli()
|
||||||
|
future := time.Now().Add(30 * 24 * time.Hour).UnixMilli()
|
||||||
|
clients := []model.Client{
|
||||||
|
{Email: "x@stale", ID: "aaaaaaaa-0000-0000-0000-00000000000a", SubID: "sub-x", Enable: true, ExpiryTime: future},
|
||||||
|
{Email: "y@stale", ID: "aaaaaaaa-0000-0000-0000-00000000000b", SubID: "sub-y", Enable: false, Reset: 30, ExpiryTime: past, TotalGB: 1000},
|
||||||
|
}
|
||||||
|
ib := &model.Inbound{
|
||||||
|
Tag: "stale-" + strconv.Itoa(port), Enable: true, Port: port, Protocol: model.VLESS,
|
||||||
|
Settings: clientsSettings(t, clients), NodeID: nodeID,
|
||||||
|
}
|
||||||
|
if err := database.GetDB().Create(ib).Error; err != nil {
|
||||||
|
t.Fatalf("create inbound: %v", err)
|
||||||
|
}
|
||||||
|
if err := (&ClientService{}).SyncInbound(nil, ib.Id, clients); err != nil {
|
||||||
|
t.Fatalf("SyncInbound: %v", err)
|
||||||
|
}
|
||||||
|
rows := []xray.ClientTraffic{
|
||||||
|
{InboundId: ib.Id, Email: "x@stale", Enable: true, ExpiryTime: future},
|
||||||
|
{InboundId: ib.Id, Email: "y@stale", Enable: false, Up: 600, Down: 400, Total: 1000, Reset: 30, ExpiryTime: past},
|
||||||
|
}
|
||||||
|
if err := database.GetDB().Create(&rows).Error; err != nil {
|
||||||
|
t.Fatalf("seed client_traffics: %v", err)
|
||||||
|
}
|
||||||
|
return ib
|
||||||
|
}
|
||||||
|
|
||||||
|
func autoRenewTick(tx *gorm.DB) error {
|
||||||
|
_, _, err := (&InboundService{}).autoRenewClients(tx, newTrafficMutationBatch())
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// renewYTick writes the renewal autoRenewClients would commit for y@stale; it
|
||||||
|
// skips clients hosted only on a node, so the node case applies it directly.
|
||||||
|
func renewYTick(inboundId int) func(tx *gorm.DB) error {
|
||||||
|
return func(tx *gorm.DB) error {
|
||||||
|
var ib model.Inbound
|
||||||
|
if err := tx.First(&ib, inboundId).Error; err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
var settings map[string]any
|
||||||
|
if err := json.Unmarshal([]byte(ib.Settings), &settings); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
for _, c := range settings["clients"].([]any) {
|
||||||
|
if m := c.(map[string]any); m["email"] == "y@stale" {
|
||||||
|
m["enable"] = true
|
||||||
|
m["expiryTime"] = time.Now().Add(30 * 24 * time.Hour).UnixMilli()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
b, err := json.MarshalIndent(settings, "", " ")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return tx.Model(&model.Inbound{}).Where("id = ?", inboundId).Update("settings", string(b)).Error
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func settingsClient(t *testing.T, inboundId int, email string) (model.Client, bool) {
|
||||||
|
t.Helper()
|
||||||
|
ib, err := (&InboundService{}).GetInbound(inboundId)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetInbound: %v", err)
|
||||||
|
}
|
||||||
|
clients, err := (&InboundService{}).GetClients(ib)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetClients: %v", err)
|
||||||
|
}
|
||||||
|
for _, c := range clients {
|
||||||
|
if c.Email == email {
|
||||||
|
return c, true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return model.Client{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
func requireNeighbourRenewed(t *testing.T, inboundId int) model.Client {
|
||||||
|
t.Helper()
|
||||||
|
y, ok := settingsClient(t, inboundId, "y@stale")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("neighbour y@stale missing from settings")
|
||||||
|
}
|
||||||
|
if now := time.Now().UnixMilli(); !y.Enable || y.ExpiryTime <= now {
|
||||||
|
t.Fatalf("renewed neighbour rolled back in settings: enable=%v expiryTime=%d (now %d)", y.Enable, y.ExpiryTime, now)
|
||||||
|
}
|
||||||
|
return y
|
||||||
|
}
|
||||||
|
|
||||||
|
type staleClientOp struct {
|
||||||
|
name string
|
||||||
|
// advancesNodeFingerprint: on a node inbound the op pushes per client and
|
||||||
|
// then advances the reconcile-skip fingerprint.
|
||||||
|
advancesNodeFingerprint bool
|
||||||
|
run func(t *testing.T, ib *model.Inbound) error
|
||||||
|
}
|
||||||
|
|
||||||
|
var staleClientOps = []staleClientOp{
|
||||||
|
{"edit", true, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
rec := lookupClientRecord(t, "x@stale")
|
||||||
|
edited := rec.ToClient()
|
||||||
|
edited.Comment = "edited"
|
||||||
|
_, err := (&ClientService{}).UpdateInboundClient(&InboundService{}, &model.Inbound{
|
||||||
|
Id: ib.Id, Settings: clientsSettings(t, []model.Client{*edited}),
|
||||||
|
}, "x@stale")
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"add", true, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
_, err := (&ClientService{}).AddInboundClient(&InboundService{}, &model.Inbound{
|
||||||
|
Id: ib.Id, Settings: clientsSettings(t, []model.Client{{Email: "z@stale", ID: "aaaaaaaa-0000-0000-0000-00000000000c", Enable: true}}),
|
||||||
|
})
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"delete", true, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
_, err := (&ClientService{}).DelInboundClientByEmail(&InboundService{}, ib.Id, "x@stale", false, true)
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"bulk detach", true, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
_, _, err := (&ClientService{}).BulkDetach(&InboundService{}, []string{"x@stale"}, []int{ib.Id})
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"bulk adjust", false, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
_, _, err := (&ClientService{}).BulkAdjust(&InboundService{}, []string{"x@stale"}, 1, 0, "", nil, "")
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"bulk delete", false, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
_, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{"x@stale"}, false)
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"bulk set enable", true, func(t *testing.T, ib *model.Inbound) error {
|
||||||
|
_, _, err := (&ClientService{}).BulkSetEnable(&InboundService{}, []string{"x@stale"}, false)
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
|
||||||
|
// Each client op reads the inbound before queueing its write; a renewal the
|
||||||
|
// traffic writer commits in between must not be reverted to enable=false.
|
||||||
|
func TestClientOpsKeepNeighbourRenewedMidOp(t *testing.T) {
|
||||||
|
for i, op := range staleClientOps {
|
||||||
|
t.Run(op.name, func(t *testing.T) {
|
||||||
|
setupBulkDB(t)
|
||||||
|
ib := seedRenewableNeighbour(t, 23101+i, nil)
|
||||||
|
commitTickBetweenReadAndWrite(t, autoRenewTick, func() {
|
||||||
|
if err := op.run(t, ib); err != nil {
|
||||||
|
t.Errorf("%s: %v", op.name, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
requireNeighbourRenewed(t, ib.Id)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// An op on the renewed client itself keeps the fields it did not change.
|
||||||
|
func TestBulkAdjustOnRenewedClientKeepsRenewal(t *testing.T) {
|
||||||
|
setupBulkDB(t)
|
||||||
|
ib := seedRenewableNeighbour(t, 23120, nil)
|
||||||
|
commitTickBetweenReadAndWrite(t, autoRenewTick, func() {
|
||||||
|
if _, _, err := (&ClientService{}).BulkAdjust(&InboundService{}, []string{"y@stale"}, 0, 500, "", nil, ""); err != nil {
|
||||||
|
t.Errorf("BulkAdjust: %v", err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
if y := requireNeighbourRenewed(t, ib.Id); y.TotalGB != 1500 {
|
||||||
|
t.Fatalf("y@stale totalGB = %d, want 1500 (the adjust itself was lost)", y.TotalGB)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The node got only the per-client push, so the skip fingerprint must not claim
|
||||||
|
// it also holds the renewal the traffic writer committed mid-op.
|
||||||
|
func TestNodeClientOpsMidRenewalStillReconcileRenewal(t *testing.T) {
|
||||||
|
for i, op := range staleClientOps {
|
||||||
|
if !op.advancesNodeFingerprint {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
t.Run(op.name, func(t *testing.T) {
|
||||||
|
setupBulkDB(t)
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_, _ = w.Write([]byte(`{"success":true}`))
|
||||||
|
}))
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
u, err := url.Parse(srv.URL)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("parse url: %v", err)
|
||||||
|
}
|
||||||
|
port, _ := strconv.Atoi(u.Port())
|
||||||
|
node := &model.Node{
|
||||||
|
Name: "stale-node", Scheme: "http", Address: u.Hostname(), Port: port, BasePath: "/",
|
||||||
|
ApiToken: "tok", Enable: true, Status: "online", AllowPrivateAddress: true,
|
||||||
|
}
|
||||||
|
if err := database.GetDB().Create(node).Error; err != nil {
|
||||||
|
t.Fatalf("create node: %v", err)
|
||||||
|
}
|
||||||
|
remote := runtime.NewRemote(node, nil)
|
||||||
|
useTestRuntimeManager(t).SetRuntimeOverride(node.Id, remote)
|
||||||
|
|
||||||
|
ib := seedRenewableNeighbour(t, 23131+i, &node.Id)
|
||||||
|
remote.AdoptInboundAlias(ib, runtime.RemoteInboundOption{Id: 7, Tag: ib.Tag})
|
||||||
|
|
||||||
|
commitTickBetweenReadAndWrite(t, renewYTick(ib.Id), func() {
|
||||||
|
if err := op.run(t, ib); err != nil {
|
||||||
|
t.Errorf("%s: %v", op.name, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
requireNeighbourRenewed(t, ib.Id)
|
||||||
|
|
||||||
|
saved, err := (&InboundService{}).GetInbound(ib.Id)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetInbound: %v", err)
|
||||||
|
}
|
||||||
|
pushed, err := remote.ReconcileInbound(context.Background(), saved, true)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ReconcileInbound: %v", err)
|
||||||
|
}
|
||||||
|
if !pushed {
|
||||||
|
t.Fatal("reconcile skipped the inbound: the node never receives y@stale's renewal")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRebaseClientSettings(t *testing.T) {
|
||||||
|
const base = `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`
|
||||||
|
cases := []struct {
|
||||||
|
name, ours, current, want string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "untouched client takes the committed version",
|
||||||
|
ours: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":2},{"email":"b","enable":false,"expiryTime":1}]}`,
|
||||||
|
current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9}]}`,
|
||||||
|
want: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":2},{"email":"b","enable":true,"expiryTime":9}]}`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "edited client keeps committed changes to fields the op left alone",
|
||||||
|
ours: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1,"comment":"x"}]}`,
|
||||||
|
current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9}]}`,
|
||||||
|
want: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9,"comment":"x"}]}`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "client the op removed stays removed",
|
||||||
|
ours: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1}]}`,
|
||||||
|
current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9}]}`,
|
||||||
|
want: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1}]}`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "client committed after the read is kept",
|
||||||
|
ours: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`,
|
||||||
|
current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1},{"email":"c"}]}`,
|
||||||
|
want: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1},{"email":"b","enable":false,"expiryTime":1},{"email":"c"}]}`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "untouched client removed after the read stays removed",
|
||||||
|
ours: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`,
|
||||||
|
current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1}]}`,
|
||||||
|
want: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1}]}`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "top-level key follows whichever side changed it",
|
||||||
|
ours: `{"decryption":"none","testseed":[1],"clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`,
|
||||||
|
current: `{"decryption":"mlkem","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`,
|
||||||
|
want: `{"decryption":"mlkem","testseed":[1],"clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
got, err := rebaseClientSettings(base, tc.ours, tc.current)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("rebaseClientSettings: %v", err)
|
||||||
|
}
|
||||||
|
var gotV, wantV any
|
||||||
|
if err := json.Unmarshal([]byte(got), &gotV); err != nil {
|
||||||
|
t.Fatalf("unmarshal got: %v", err)
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal([]byte(tc.want), &wantV); err != nil {
|
||||||
|
t.Fatalf("unmarshal want: %v", err)
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(gotV, wantV) {
|
||||||
|
t.Fatalf("rebase = %s\nwant %s", got, tc.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user