fix(onlines): 恢复被 P9 掏空的在线状态存储(第11条)

根因: 1035d00 删本地核心时, inbound_node.go 的在线存储层被一并掏空
(GetOnlineClients 写死空/SetNodeOnlineTree 等空函数/SetRemoteTraffic
末尾喂树整块删除), 而节点轮询与 agent 上报两条到达路径完好, 数据
送达即被扔, 客户端/入站在线永远全零。

修复: 把原 xray.Process 上的在线存储(算法/锁/grace 20s 语义逐字
移植)落到 service 包单例 onlineStore(sharedOnlineStore, 对应原
currentXrayProcess 的单一持有语义); 恢复六个 wrapper + 新增
SetNodeActiveInboundTree; setRemoteTrafficLocked 补回 guidShared
与喂树块(原 f727d04 逻辑, process. 换 s.)。

偏离: 本地集在精简主控无本地核心喂养, 仅 mtproto job 折入(该 job
已不被调度, 路径保留无害); GetOnlineClients 对 nil 归一为 []。

测试: online_store_test 9 例(-race 全过: 并集/按GUID/替换/清空/
grace 边界/跨实例共享/并发); SetRemoteTraffic 集成 2 例(传统节点
GUID 树+active 树落库、旧节点 flat 回退); 全量 go test 串行复跑失败集合与基线 9 红逐字一致(零新增);
service 包单跑唯一红为基线既有的 TestGetXrayConfig 红, 其余全过。
This commit is contained in:
Muse
2026-10-22 09:19:20 +00:00
parent f569680182
commit ac39370879
4 changed files with 683 additions and 10 deletions
+66 -10
View File
@@ -439,6 +439,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
nodeRow.Id = nodeID nodeRow.Id = nodeID
unmanagedTag := unmanagedTagPredicate(&nodeRow) unmanagedTag := unmanagedTagPredicate(&nodeRow)
selfKey := effectiveNodeKey(&model.Node{Id: nodeID, Guid: nodeRow.Guid}) selfKey := effectiveNodeKey(&model.Node{Id: nodeID, Guid: nodeRow.Guid})
guidShared := nodeRow.Guid != "" && selfKey != nodeRow.Guid
originGuidFor := func(snapIb *model.Inbound) string { originGuidFor := func(snapIb *model.Inbound) string {
if snapIb.OriginNodeGuid != "" && snapIb.OriginNodeGuid != nodeRow.Guid { if snapIb.OriginNodeGuid != "" && snapIb.OriginNodeGuid != nodeRow.Guid {
return snapIb.OriginNodeGuid return snapIb.OriginNodeGuid
@@ -1267,6 +1268,32 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
} }
} }
tree := snap.OnlineTree
switch {
case len(tree) == 0 && len(snap.OnlineEmails) > 0:
// Old-build node (no GUID tree): key its flat online list under its
// own effective identity so attribution still works for that branch.
tree = map[string][]string{selfKey: snap.OnlineEmails}
case guidShared && len(tree) > 0:
// Newer cloned node: its own clients arrive keyed under the shared
// panelGuid. Remap just that entry to the node-unique key so the
// clones don't merge; descendant subtrees keep their distinct GUIDs.
if _, ok := tree[nodeRow.Guid]; ok {
tree = remapGuidTreeKey(tree, nodeRow.Guid, selfKey)
}
}
s.SetNodeOnlineTree(nodeID, tree)
activeTree := normalizeActiveInboundTreeTags(snap.ActiveInboundTree, tagToCentral)
if guidShared && len(activeTree) > 0 {
if _, ok := activeTree[nodeRow.Guid]; ok {
activeTree = remapGuidTreeKey(activeTree, nodeRow.Guid, selfKey)
}
}
if len(activeTree) > 0 {
activeTree = filterGuidTreeKeys(activeTree, activeInboundGuidKeys(snap.Inbounds, tagToCentral, originGuidFor))
}
s.SetNodeActiveInboundTree(nodeID, activeTree)
return structuralChange, nil return structuralChange, nil
} }
@@ -1294,8 +1321,11 @@ func (s *InboundService) restartRemoteNodesOnDisable(nodeIDs []int) {
} }
func (s *InboundService) GetOnlineClients() []string { func (s *InboundService) GetOnlineClients() []string {
// Lite master: no local core aggregates online clients; node onlines // Lite master: no local core aggregates online clients; the online
// arrive through agent reports and node polling instead. // store (see online_store.go) is fed by agent reports and node polling.
if out := sharedOnlineStore.GetOnlineClients(); out != nil {
return out
}
return []string{} return []string{}
} }
@@ -1305,7 +1335,13 @@ func (s *InboundService) GetOnlineClients() []string {
// node-id keying so a client three hops down is attributed to its real node, // node-id keying so a client three hops down is attributed to its real node,
// not the intermediate one it was synced through. // not the intermediate one it was synced through.
func (s *InboundService) GetOnlineClientsByGuid() map[string][]string { func (s *InboundService) GetOnlineClientsByGuid() map[string][]string {
return map[string][]string{} out := sharedOnlineStore.GetMergedNodeTrees()
if local := sharedOnlineStore.GetLocalOnlineClients(); len(local) > 0 {
if guid := s.panelGuid(); guid != "" {
out[guid] = mergeEmails(out[guid], local)
}
}
return out
} }
// GetActiveInboundsByGuid returns the inbound tags that carried traffic within // GetActiveInboundsByGuid returns the inbound tags that carried traffic within
@@ -1313,12 +1349,30 @@ func (s *InboundService) GetOnlineClientsByGuid() map[string][]string {
// each inbound. A GUID missing from the map means "don't gate" for that node's // each inbound. A GUID missing from the map means "don't gate" for that node's
// inbounds (old-build node or no active-inbound signal). // inbounds (old-build node or no active-inbound signal).
func (s *InboundService) GetActiveInboundsByGuid() map[string][]string { func (s *InboundService) GetActiveInboundsByGuid() map[string][]string {
return map[string][]string{} out := sharedOnlineStore.GetMergedActiveInboundTrees()
active := sharedOnlineStore.GetLocalActiveInbounds()
if len(active) == 0 {
return out
}
guid := s.panelGuid()
if guid == "" {
return out
}
out[guid] = mergeEmails(out[guid], active)
return out
} }
func (s *InboundService) SetNodeOnlineTree(nodeID int, tree map[string][]string) {} func (s *InboundService) SetNodeOnlineTree(nodeID int, tree map[string][]string) {
sharedOnlineStore.SetNodeOnlineTree(nodeID, tree)
}
func (s *InboundService) ClearNodeOnlineClients(nodeID int) {} func (s *InboundService) SetNodeActiveInboundTree(nodeID int, tree map[string][]string) {
sharedOnlineStore.SetNodeActiveInboundTree(nodeID, tree)
}
func (s *InboundService) ClearNodeOnlineClients(nodeID int) {
sharedOnlineStore.ClearNodeOnlineClients(nodeID)
}
// panelGuid returns this panel's stable self-identifier, used to key the local // panelGuid returns this panel's stable self-identifier, used to key the local
// panel's own clients in the per-node online maps (#4983). // panel's own clients in the per-node online maps (#4983).
@@ -1451,11 +1505,13 @@ func (s *InboundService) GetClientsLastOnline() (map[string]int64, error) {
} }
// RefreshLocalOnlineClients folds the emails and inbound tags active on this // RefreshLocalOnlineClients folds the emails and inbound tags active on this
// panel's own xray this poll into the local online/active sets, applying the // panel's own inbounds this poll into the local online/active sets, applying
// grace window and pruning stale entries. Pass nil to only prune. See // the grace window and pruning stale entries. Pass nil to only prune. See
// xray.Process for why the local sets are kept separate from the shared // onlineStore for why the local sets are kept separate from the shared
// last_online column. // last_online column.
func (s *InboundService) RefreshLocalOnlineClients(activeEmails, activeInboundTags []string) {} func (s *InboundService) RefreshLocalOnlineClients(activeEmails, activeInboundTags []string) {
sharedOnlineStore.RefreshLocalOnline(activeEmails, activeInboundTags, time.Now().UnixMilli(), onlineGracePeriodMs)
}
func (s *InboundService) FilterAndSortClientEmails(emails []string) ([]string, []string, error) { func (s *InboundService) FilterAndSortClientEmails(emails []string) ([]string, []string, error) {
db := database.GetDB() db := database.GetDB()
@@ -0,0 +1,122 @@
package service
import (
"testing"
"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"
)
// The traditional-node path: NodeTrafficSyncJob fetches a TrafficSnapshot
// from each node panel (runtime.Remote.FetchTrafficSnapshot) and hands it
// to InboundService.SetRemoteTraffic. Before the online-store restoration,
// the snapshot's OnlineTree / ActiveInboundTree were dropped on the floor
// inside that call (the store setters were no-op stubs), so polled nodes
// never showed anyone online. This drives the exact public entry the job
// calls and asserts the snapshot's online data lands in the store.
func TestSetRemoteTraffic_FeedsOnlineStore(t *testing.T) {
setupConflictDB(t)
db := database.GetDB()
svc := InboundService{}
const nodeID = 4242
const nodeGuid = "trad-node-guid-4242"
svc.ClearNodeOnlineClients(nodeID)
t.Cleanup(func() { svc.ClearNodeOnlineClients(nodeID) })
if err := db.Create(&model.Node{
Id: nodeID,
Name: "trad-node",
Address: "10.0.4.2",
Port: 2053,
ApiToken: "t",
Guid: nodeGuid,
}).Error; err != nil {
t.Fatalf("create node: %v", err)
}
snap := &runtime.TrafficSnapshot{
Inbounds: []*model.Inbound{
{
Tag: "in-443-tcp",
Enable: true,
Port: 443,
Protocol: model.VLESS,
Settings: `{"clients":[{"email":"trad@example.com"}]}`,
},
},
OnlineTree: map[string][]string{nodeGuid: {"trad@example.com"}},
ActiveInboundTree: map[string][]string{nodeGuid: {"in-443-tcp"}},
}
if _, err := svc.SetRemoteTraffic(nodeID, snap, false, false); err != nil {
t.Fatalf("SetRemoteTraffic: %v", err)
}
assertStringSet(t, "online union", svc.GetOnlineClients(), []string{"trad@example.com"})
byGuid := svc.GetOnlineClientsByGuid()
assertStringSet(t, "online byGuid", byGuid[nodeGuid], []string{"trad@example.com"})
// The active tree is normalized from node-side tags to central tags;
// whatever tag the adopted central inbound carries is what must show.
var central model.Inbound
if err := db.Where("node_id = ?", nodeID).First(&central).Error; err != nil {
t.Fatalf("load central inbound: %v", err)
}
active := svc.GetActiveInboundsByGuid()
assertStringSet(t, "active byGuid", active[nodeGuid], []string{central.Tag})
// A failed probe makes the job clear the node; its clients must stop
// showing as online immediately.
svc.ClearNodeOnlineClients(nodeID)
if got := svc.GetOnlineClients(); len(got) != 0 {
t.Fatalf("online union after clear = %v, want empty", got)
}
}
// Old-build nodes have no GUID tree endpoint: the job passes their flat
// OnlineEmails in the snapshot and SetRemoteTraffic must key them under
// the node's own effective identity.
func TestSetRemoteTraffic_FlatOnlineEmailsFallback(t *testing.T) {
setupConflictDB(t)
db := database.GetDB()
svc := InboundService{}
const nodeID = 4243
const nodeGuid = "flat-node-guid-4243"
svc.ClearNodeOnlineClients(nodeID)
t.Cleanup(func() { svc.ClearNodeOnlineClients(nodeID) })
if err := db.Create(&model.Node{
Id: nodeID,
Name: "flat-node",
Address: "10.0.4.3",
Port: 2053,
ApiToken: "t",
Guid: nodeGuid,
}).Error; err != nil {
t.Fatalf("create node: %v", err)
}
snap := &runtime.TrafficSnapshot{
Inbounds: []*model.Inbound{
{
Tag: "in-8443-tcp",
Enable: true,
Port: 8443,
Protocol: model.VLESS,
Settings: `{"clients":[{"email":"flat@example.com"}]}`,
},
},
OnlineEmails: []string{"flat@example.com"},
}
if _, err := svc.SetRemoteTraffic(nodeID, snap, false, false); err != nil {
t.Fatalf("SetRemoteTraffic: %v", err)
}
assertStringSet(t, "online union (flat)", svc.GetOnlineClients(), []string{"flat@example.com"})
byGuid := svc.GetOnlineClientsByGuid()
assertStringSet(t, "online byGuid (flat)", byGuid[nodeGuid], []string{"flat@example.com"})
}
+302
View File
@@ -0,0 +1,302 @@
package service
import "sync"
// onlineStore is the panel's in-memory online-clients store: which client
// emails are currently online, and on which inbound tags, keyed by the
// panelGuid of the node that physically hosts them (#4983).
//
// This is the restored pre-P9 implementation. The store originally lived on
// xray.Process (the local core handle) and the service wrappers reached it
// through currentXrayProcess(); P9 deleted the local core — and with it the
// only Process instance — but kept every data-arrival path (node polling,
// agent reports) calling the service setters, which were left as no-op
// stubs, so online state silently stayed empty. The lite master has no
// local core to hang the store on, so it now lives here as a package-level
// singleton shared by every InboundService value (services are constructed
// as bare values in many places, so an instance field would fragment the
// state between writers — agent controller, sync job — and readers —
// client controller, paging service). The algorithms, locking and
// grace-window semantics are ported verbatim from xray.Process.
//
// The local sets (onlineClients / localActiveInbounds) track THIS panel's
// own inbounds only. On the lite master no local core feeds them — only
// RefreshLocalOnlineClients fold-ins (e.g. the mtproto job's local actives)
// — while node sets arrive per node through the Set*Tree setters. The local
// sets are kept independent of the shared client_traffics.last_online
// column: that column is bumped by remote-node syncs too and would
// otherwise leak remote-only clients into the local set.
type onlineStore struct {
// onlineClients is the set of emails active on THIS panel's own
// inbounds within the online grace window. It is derived only from
// local fold-ins (see RefreshLocalOnline) — never from remote-node
// snapshots — so a client connected solely to a remote node is not
// reported online on local inbounds.
onlineClients []string
// localActiveInbounds is the set of THIS panel's inbound tags that
// carried traffic within the same grace window. A user>>>email stat
// aggregates across every inbound a client is attached to, so an
// online email alone can't say which inbound it actually used. Pairing
// it with the inbound>>>tag stat lets the per-inbound view drop a
// multi-inbound client from inbounds that saw no traffic this window.
localActiveInbounds []string
// localLastOnline records, per email, the last time a local fold-in
// reported traffic for it. RefreshLocalOnline rebuilds onlineClients
// from this map each tick.
localLastOnline map[string]int64
// localInboundLastActive mirrors localLastOnline for inbound tags: the
// last tick traffic was reported through each tag. Rebuilt into
// localActiveInbounds under the same grace window so the two signals
// stay aligned — an email within grace always has the inbound it used
// within grace too.
localInboundLastActive map[string]int64
// nodeOnlineTrees holds, per direct remote node (keyed by that node's
// panel-local id), the GUID-keyed online-emails subtree that node
// reported — its own clients under its panelGuid plus every descendant
// under theirs. Keying the stored value by GUID (not node id) lets the
// master attribute a deeply nested client to the node that physically
// hosts it across a chain (#4983); the outer node-id key is only so a
// failed probe can drop that whole branch's contribution.
// NodeTrafficSyncJob populates entries per cron tick and clears them
// when a probe fails. The mutex guards this map, onlineClients, and
// localLastOnline above so the online getters never see a torn read.
nodeOnlineTrees map[int]map[string][]string
// nodeActiveInboundTrees mirrors nodeOnlineTrees for active inbound
// tags: each direct node reports a GUID-keyed subtree of inbound tags
// that carried traffic within its own grace window. The inbounds page
// combines this with nodeOnlineTrees so a multi-inbound client is not
// shown on an idle inbound.
nodeActiveInboundTrees map[int]map[string][]string
mu sync.RWMutex
}
// sharedOnlineStore is the one online store for the whole panel process —
// the counterpart of the single xray.Process the pre-P9 code reached via
// currentXrayProcess().
var sharedOnlineStore = &onlineStore{}
// GetOnlineClients returns the union of locally-online clients and
// node-online clients from every registered remote panel. Dedupes by
// email so a client connected to both a local and a node-managed inbound
// surfaces once. Cheap allocation — typical online sets are small and
// the union is recomputed on demand.
func (p *onlineStore) GetOnlineClients() []string {
p.mu.RLock()
defer p.mu.RUnlock()
if len(p.nodeOnlineTrees) == 0 {
// Hot path for single-panel deployments: avoid the map+dedupe
// work entirely and return the local slice as-is.
return p.onlineClients
}
seen := make(map[string]struct{}, len(p.onlineClients))
out := make([]string, 0, len(p.onlineClients))
add := func(emails []string) {
for _, email := range emails {
if _, dup := seen[email]; dup {
continue
}
seen[email] = struct{}{}
out = append(out, email)
}
}
add(p.onlineClients)
for _, tree := range p.nodeOnlineTrees {
for _, emails := range tree {
add(emails)
}
}
return out
}
// GetLocalOnlineClients returns a copy of the emails online on THIS panel's
// own inbounds within the grace window. The service layer keys these under
// the panel's own GUID when assembling the per-node online view.
func (p *onlineStore) GetLocalOnlineClients() []string {
p.mu.RLock()
defer p.mu.RUnlock()
if len(p.onlineClients) == 0 {
return nil
}
out := make([]string, len(p.onlineClients))
copy(out, p.onlineClients)
return out
}
// GetMergedNodeTrees returns the union of every direct node's reported
// subtree, keyed by the panelGuid of the node that physically hosts each
// client set. Because each child already reports its descendants under
// their own GUIDs, merging the direct children yields the whole tree at
// any depth (#4983), so a client three hops down is attributed to its real
// node, not the intermediate one. GUIDs are globally unique, but a set
// reported under the same GUID by more than one path is deduped per key;
// empty sets are omitted.
func (p *onlineStore) GetMergedNodeTrees() map[string][]string {
p.mu.RLock()
defer p.mu.RUnlock()
if len(p.nodeOnlineTrees) == 0 {
return map[string][]string{}
}
out := make(map[string][]string)
seen := make(map[string]map[string]struct{})
for _, tree := range p.nodeOnlineTrees {
for guid, emails := range tree {
if guid == "" || len(emails) == 0 {
continue
}
dedup := seen[guid]
if dedup == nil {
dedup = make(map[string]struct{}, len(emails))
seen[guid] = dedup
}
for _, email := range emails {
if _, ok := dedup[email]; ok {
continue
}
dedup[email] = struct{}{}
out[guid] = append(out[guid], email)
}
}
}
return out
}
// GetLocalActiveInbounds returns a copy of THIS panel's inbound tags that
// carried traffic within the grace window. The service layer keys these
// under the panel's own GUID before merging them with remote-node
// active-inbound subtrees.
func (p *onlineStore) GetLocalActiveInbounds() []string {
p.mu.RLock()
defer p.mu.RUnlock()
if len(p.localActiveInbounds) == 0 {
return nil
}
out := make([]string, len(p.localActiveInbounds))
copy(out, p.localActiveInbounds)
return out
}
// GetMergedActiveInboundTrees returns the union of every direct node's
// reported active-inbound subtree, keyed by the panelGuid of the node that
// physically hosts each inbound. Duplicate tags reported through multiple
// paths are deduped per GUID.
func (p *onlineStore) GetMergedActiveInboundTrees() map[string][]string {
p.mu.RLock()
defer p.mu.RUnlock()
if len(p.nodeActiveInboundTrees) == 0 {
return map[string][]string{}
}
out := make(map[string][]string)
seen := make(map[string]map[string]struct{})
for _, tree := range p.nodeActiveInboundTrees {
for guid, tags := range tree {
if guid == "" || len(tags) == 0 {
continue
}
dedup := seen[guid]
if dedup == nil {
dedup = make(map[string]struct{}, len(tags))
seen[guid] = dedup
}
for _, tag := range tags {
if tag == "" {
continue
}
if _, ok := dedup[tag]; ok {
continue
}
dedup[tag] = struct{}{}
out[guid] = append(out[guid], tag)
}
}
}
return out
}
// RefreshLocalOnline records that each email in activeEmails and each tag
// in activeInboundTags had local traffic at now, then rebuilds
// onlineClients and localActiveInbounds from every entry seen within
// graceMs, pruning older ones. Pass nil/empty slices to only prune —
// NodeTrafficSyncJob does this so stale local entries still age out
// between fold-ins.
func (p *onlineStore) RefreshLocalOnline(activeEmails, activeInboundTags []string, now, graceMs int64) {
p.mu.Lock()
defer p.mu.Unlock()
if p.localLastOnline == nil {
p.localLastOnline = make(map[string]int64, len(activeEmails))
}
for _, email := range activeEmails {
p.localLastOnline[email] = now
}
online := make([]string, 0, len(p.localLastOnline))
for email, ts := range p.localLastOnline {
if now-ts < graceMs {
online = append(online, email)
} else {
delete(p.localLastOnline, email)
}
}
p.onlineClients = online
if p.localInboundLastActive == nil {
p.localInboundLastActive = make(map[string]int64, len(activeInboundTags))
}
for _, tag := range activeInboundTags {
p.localInboundLastActive[tag] = now
}
activeInbounds := make([]string, 0, len(p.localInboundLastActive))
for tag, ts := range p.localInboundLastActive {
if now-ts < graceMs {
activeInbounds = append(activeInbounds, tag)
} else {
delete(p.localInboundLastActive, tag)
}
}
p.localActiveInbounds = activeInbounds
}
// SetNodeOnlineTree records the GUID-keyed online subtree one direct remote
// node reported (its own clients under its panelGuid plus every descendant
// under theirs). Replaces any previous entry for that node —
// NodeTrafficSyncJob always sends the full subtree per tick, and the agent
// controller sends its online list per traffic report.
func (p *onlineStore) SetNodeOnlineTree(nodeID int, tree map[string][]string) {
p.mu.Lock()
defer p.mu.Unlock()
if len(tree) == 0 {
delete(p.nodeOnlineTrees, nodeID)
return
}
if p.nodeOnlineTrees == nil {
p.nodeOnlineTrees = map[int]map[string][]string{}
}
p.nodeOnlineTrees[nodeID] = tree
}
// SetNodeActiveInboundTree records the GUID-keyed active-inbound subtree
// one direct remote node reported. Replaces any previous entry for that
// node.
func (p *onlineStore) SetNodeActiveInboundTree(nodeID int, tree map[string][]string) {
p.mu.Lock()
defer p.mu.Unlock()
if len(tree) == 0 {
delete(p.nodeActiveInboundTrees, nodeID)
return
}
if p.nodeActiveInboundTrees == nil {
p.nodeActiveInboundTrees = map[int]map[string][]string{}
}
p.nodeActiveInboundTrees[nodeID] = tree
}
// ClearNodeOnlineClients drops a direct node's whole subtree contribution
// (online emails and active inbounds). Called when a probe fails or an
// agent disconnects, so a downed node — and everything behind it — doesn't
// keep its clients listed as "online" until the next successful probe.
func (p *onlineStore) ClearNodeOnlineClients(nodeID int) {
p.mu.Lock()
defer p.mu.Unlock()
delete(p.nodeOnlineTrees, nodeID)
delete(p.nodeActiveInboundTrees, nodeID)
}
+193
View File
@@ -0,0 +1,193 @@
package service
import (
"fmt"
"sync"
"testing"
"time"
)
func sortedSet(in []string) map[string]bool {
out := make(map[string]bool, len(in))
for _, s := range in {
out[s] = true
}
return out
}
func requireSet(t *testing.T, what string, got []string, want ...string) {
t.Helper()
if len(got) != len(want) {
t.Fatalf("%s: got %v, want set %v", what, got, want)
}
set := sortedSet(got)
for _, w := range want {
if !set[w] {
t.Fatalf("%s: got %v, missing %q (want %v)", what, got, w, want)
}
}
}
// A node's reported tree must surface both in the flat union and in the
// GUID-keyed view, deduped against the local set and other nodes.
func TestOnlineStoreSetTreeUnionAndByGuid(t *testing.T) {
st := &onlineStore{}
st.SetNodeOnlineTree(1, map[string][]string{
"guid-a": {"alice@example.com", "bob@example.com"},
"guid-b": {"carol@example.com"},
})
st.SetNodeOnlineTree(2, map[string][]string{
"guid-c": {"bob@example.com", "dave@example.com"},
})
// Local fold-in overlapping one node email (dedupe check).
st.RefreshLocalOnline([]string{"alice@example.com", "local@example.com"}, nil, 1000, 20000)
requireSet(t, "union", st.GetOnlineClients(),
"alice@example.com", "bob@example.com", "carol@example.com", "dave@example.com", "local@example.com")
merged := st.GetMergedNodeTrees()
requireSet(t, "byGuid guid-a", merged["guid-a"], "alice@example.com", "bob@example.com")
requireSet(t, "byGuid guid-b", merged["guid-b"], "carol@example.com")
requireSet(t, "byGuid guid-c", merged["guid-c"], "bob@example.com", "dave@example.com")
if _, ok := merged["guid-absent"]; ok {
t.Fatalf("byGuid: unexpected key for absent guid: %v", merged)
}
requireSet(t, "local", st.GetLocalOnlineClients(), "alice@example.com", "local@example.com")
}
// The agent controller reports onlines under the literal "agent" GUID key;
// they must reach the union untouched.
func TestOnlineStoreAgentKeyTree(t *testing.T) {
st := &onlineStore{}
st.SetNodeOnlineTree(7, map[string][]string{"agent": {"agentuser@example.com"}})
requireSet(t, "union", st.GetOnlineClients(), "agentuser@example.com")
requireSet(t, "byGuid agent", st.GetMergedNodeTrees()["agent"], "agentuser@example.com")
}
// Setting a tree replaces the node's previous contribution; an empty tree
// deletes it outright.
func TestOnlineStoreSetTreeReplacesAndEmptyDeletes(t *testing.T) {
st := &onlineStore{}
st.SetNodeOnlineTree(3, map[string][]string{"g": {"one@example.com"}})
st.SetNodeOnlineTree(3, map[string][]string{"g": {"two@example.com"}})
requireSet(t, "union after replace", st.GetOnlineClients(), "two@example.com")
st.SetNodeOnlineTree(3, map[string][]string{})
if got := st.GetOnlineClients(); len(got) != 0 {
t.Fatalf("union after empty set: got %v, want empty", got)
}
if got := st.GetMergedNodeTrees(); len(got) != 0 {
t.Fatalf("byGuid after empty set: got %v, want empty", got)
}
}
// Active-inbound trees behave like online trees, with per-GUID dedupe.
func TestOnlineStoreActiveInboundTrees(t *testing.T) {
st := &onlineStore{}
st.SetNodeActiveInboundTree(1, map[string][]string{"g1": {"in-a", "in-b"}})
st.SetNodeActiveInboundTree(2, map[string][]string{"g1": {"in-b", "in-c"}})
merged := st.GetMergedActiveInboundTrees()
requireSet(t, "active g1", merged["g1"], "in-a", "in-b", "in-c")
st.RefreshLocalOnline(nil, []string{"local-in"}, 5000, 20000)
requireSet(t, "local active", st.GetLocalActiveInbounds(), "local-in")
}
// Clear drops BOTH of the node's trees and leaves other nodes untouched.
func TestOnlineStoreClear(t *testing.T) {
st := &onlineStore{}
st.SetNodeOnlineTree(1, map[string][]string{"g1": {"a@example.com"}})
st.SetNodeActiveInboundTree(1, map[string][]string{"g1": {"in-a"}})
st.SetNodeOnlineTree(2, map[string][]string{"g2": {"b@example.com"}})
st.ClearNodeOnlineClients(1)
requireSet(t, "union after clear", st.GetOnlineClients(), "b@example.com")
if got := st.GetMergedActiveInboundTrees(); len(got) != 0 {
t.Fatalf("active after clear: got %v, want empty", got)
}
}
// Local entries live for the grace window, then a prune-only refresh
// (nil, nil — what NodeTrafficSyncJob calls every tick) ages them out.
func TestOnlineStoreGraceWindowPrune(t *testing.T) {
st := &onlineStore{}
const grace = int64(20000)
st.RefreshLocalOnline([]string{"stay@example.com"}, []string{"tag-stay"}, 10000, grace)
st.RefreshLocalOnline([]string{"fresh@example.com"}, nil, 25000, grace)
// At t=25000 the first entry is 15s old: still inside the window.
requireSet(t, "local within grace", st.GetLocalOnlineClients(), "stay@example.com", "fresh@example.com")
// Prune-only at t=45001: stay@10000 is 35s old (gone), fresh@25000 is
// 20001ms old (also past the strict < grace comparison), tag pruned.
st.RefreshLocalOnline(nil, nil, 45001, grace)
if got := st.GetLocalOnlineClients(); len(got) != 0 {
t.Fatalf("local after prune: got %v, want empty", got)
}
if got := st.GetLocalActiveInbounds(); len(got) != 0 {
t.Fatalf("local active after prune: got %v, want empty", got)
}
}
// Boundary: an entry exactly graceMs old is expired (original uses <).
func TestOnlineStoreGraceBoundary(t *testing.T) {
st := &onlineStore{}
st.RefreshLocalOnline([]string{"edge@example.com"}, nil, 0, 20000)
st.RefreshLocalOnline(nil, nil, 19999, 20000)
requireSet(t, "just inside", st.GetLocalOnlineClients(), "edge@example.com")
st.RefreshLocalOnline(nil, nil, 20000, 20000)
if got := st.GetLocalOnlineClients(); len(got) != 0 {
t.Fatalf("at boundary: got %v, want empty", got)
}
}
// The store is shared across InboundService values: a write through one
// instance must be visible through another (services are bare values; an
// instance field would fragment state between writers and readers).
func TestOnlineStoreSharedAcrossServiceInstances(t *testing.T) {
writer := &InboundService{}
reader := &InboundService{}
const nodeID = 987654
t.Cleanup(func() { writer.ClearNodeOnlineClients(nodeID) })
writer.SetNodeOnlineTree(nodeID, map[string][]string{"guid-shared": {"shared@example.com"}})
requireSet(t, "union via other instance", reader.GetOnlineClients(), "shared@example.com")
byGuid := reader.GetOnlineClientsByGuid()
requireSet(t, "byGuid via other instance", byGuid["guid-shared"], "shared@example.com")
}
// Concurrent writers/readers must not race or corrupt the union. Run with
// -race for the full guarantee; even without it, the final state is exact.
func TestOnlineStoreConcurrentAccess(t *testing.T) {
st := &onlineStore{}
const nodes = 8
var wg sync.WaitGroup
for n := 0; n < nodes; n++ {
n := n
wg.Add(1)
go func() {
defer wg.Done()
guid := fmt.Sprintf("guid-%d", n)
for i := 0; i < 50; i++ {
st.SetNodeOnlineTree(n, map[string][]string{guid: {fmt.Sprintf("u%d-%d@example.com", n, i)}})
_ = st.GetOnlineClients()
_ = st.GetMergedNodeTrees()
st.RefreshLocalOnline([]string{fmt.Sprintf("local-%d@example.com", n)}, nil, time.Now().UnixMilli(), 60000)
}
}()
}
wg.Wait()
// Every node's last write holds exactly one email; locals may or may
// not have been pruned by a same-ms refresh, so assert on node trees.
merged := st.GetMergedNodeTrees()
for n := 0; n < nodes; n++ {
requireSet(t, fmt.Sprintf("tree %d", n), merged[fmt.Sprintf("guid-%d", n)],
fmt.Sprintf("u%d-49@example.com", n))
}
for n := 0; n < nodes; n++ {
st.ClearNodeOnlineClients(n)
}
if got := st.GetMergedNodeTrees(); len(got) != 0 {
t.Fatalf("trees after mass clear: got %v, want empty", got)
}
}