feat(node/tracker): implement initial refresh gate and bootstrap completion logic
This commit is contained in:
@@ -76,6 +76,13 @@ type Tracker struct {
|
|||||||
uuidToName map[string]string // uuid → name (from node list)
|
uuidToName map[string]string // uuid → name (from node list)
|
||||||
onlineSet map[string]bool // which uuids are currently online
|
onlineSet map[string]bool // which uuids are currently online
|
||||||
callbacks []StatusChangeCallback
|
callbacks []StatusChangeCallback
|
||||||
|
|
||||||
|
// Initial-refresh gating: status-change notifications are suppressed until
|
||||||
|
// every node from the Komari node list has delivered a status update, so the
|
||||||
|
// startup state is treated as a baseline rather than a flood of changes.
|
||||||
|
knownUUIDs map[string]bool // uuids from the Komari node list
|
||||||
|
reportedUUIDs map[string]bool // node-list uuids that have reported status
|
||||||
|
bootstrapDone bool // true once the initial refresh has finished
|
||||||
}
|
}
|
||||||
|
|
||||||
var globalTracker *Tracker
|
var globalTracker *Tracker
|
||||||
@@ -86,6 +93,8 @@ func InitTracker() {
|
|||||||
nodes: make(map[string]*Node),
|
nodes: make(map[string]*Node),
|
||||||
uuidToName: make(map[string]string),
|
uuidToName: make(map[string]string),
|
||||||
onlineSet: make(map[string]bool),
|
onlineSet: make(map[string]bool),
|
||||||
|
knownUUIDs: make(map[string]bool),
|
||||||
|
reportedUUIDs: make(map[string]bool),
|
||||||
}
|
}
|
||||||
postLog.Info("Node tracker initialized")
|
postLog.Info("Node tracker initialized")
|
||||||
}
|
}
|
||||||
@@ -121,6 +130,13 @@ func (t *Tracker) UpdateNodeList(nodeNames map[string]string) {
|
|||||||
t.mu.Lock()
|
t.mu.Lock()
|
||||||
defer t.mu.Unlock()
|
defer t.mu.Unlock()
|
||||||
|
|
||||||
|
// The node list defines the set of nodes to wait for during the initial
|
||||||
|
// status refresh.
|
||||||
|
t.knownUUIDs = make(map[string]bool, len(nodeNames))
|
||||||
|
for uuid := range nodeNames {
|
||||||
|
t.knownUUIDs[uuid] = true
|
||||||
|
}
|
||||||
|
|
||||||
t.uuidToName = make(map[string]string, len(nodeNames))
|
t.uuidToName = make(map[string]string, len(nodeNames))
|
||||||
for uuid, name := range nodeNames {
|
for uuid, name := range nodeNames {
|
||||||
t.uuidToName[uuid] = name
|
t.uuidToName[uuid] = name
|
||||||
@@ -211,10 +227,35 @@ func (t *Tracker) UpdateStatus(onlineUUIDs []string, reports map[string]Report)
|
|||||||
}
|
}
|
||||||
|
|
||||||
t.onlineSet = newOnlineSet
|
t.onlineSet = newOnlineSet
|
||||||
|
|
||||||
|
// Track which node-list nodes have delivered a status update. The initial
|
||||||
|
// refresh is considered complete once every known node has reported at
|
||||||
|
// least once (either online or offline).
|
||||||
|
for _, uuid := range onlineUUIDs {
|
||||||
|
if t.knownUUIDs[uuid] {
|
||||||
|
t.reportedUUIDs[uuid] = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for uuid := range reports {
|
||||||
|
if t.knownUUIDs[uuid] {
|
||||||
|
t.reportedUUIDs[uuid] = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Changes detected while the initial refresh is still in progress represent
|
||||||
|
// the nodes' startup state rather than real transitions, so suppress them.
|
||||||
|
suppressChanges := !t.bootstrapDone
|
||||||
|
if !t.bootstrapDone && t.allKnownNodesReported() {
|
||||||
|
t.bootstrapDone = true
|
||||||
|
postLog.Info(fmt.Sprintf("All %d nodes refreshed their status; status notifications enabled", len(t.knownUUIDs)))
|
||||||
|
}
|
||||||
t.mu.Unlock()
|
t.mu.Unlock()
|
||||||
|
|
||||||
// Fire callbacks for each change (outside the lock).
|
// Fire callbacks for each change (outside the lock).
|
||||||
for _, change := range changes {
|
for _, change := range changes {
|
||||||
|
if suppressChanges {
|
||||||
|
continue
|
||||||
|
}
|
||||||
postLog.Info(fmt.Sprintf("Node status change: %s [%s] - %s", change.Name, change.UUID, change.Event))
|
postLog.Info(fmt.Sprintf("Node status change: %s [%s] - %s", change.Name, change.UUID, change.Event))
|
||||||
t.fireCallbacks(change)
|
t.fireCallbacks(change)
|
||||||
}
|
}
|
||||||
@@ -222,6 +263,29 @@ func (t *Tracker) UpdateStatus(onlineUUIDs []string, reports map[string]Report)
|
|||||||
return changes
|
return changes
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// allKnownNodesReported returns true when every node from the Komari node list
|
||||||
|
// has delivered at least one status update. Must be called with the lock held.
|
||||||
|
func (t *Tracker) allKnownNodesReported() bool {
|
||||||
|
for uuid := range t.knownUUIDs {
|
||||||
|
if !t.reportedUUIDs[uuid] {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// CompleteBootstrap force-completes the initial status refresh gate. It is a
|
||||||
|
// safety net so that notifications cannot be blocked forever by a node that
|
||||||
|
// never reports a status. Calling it again is a no-op.
|
||||||
|
func (t *Tracker) CompleteBootstrap() {
|
||||||
|
t.mu.Lock()
|
||||||
|
defer t.mu.Unlock()
|
||||||
|
if !t.bootstrapDone {
|
||||||
|
t.bootstrapDone = true
|
||||||
|
postLog.Info("Initial node status refresh timeout reached; status notifications enabled")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// GetNode returns a copy of the node data for the given UUID.
|
// GetNode returns a copy of the node data for the given UUID.
|
||||||
func (t *Tracker) GetNode(uuid string) (*Node, bool) {
|
func (t *Tracker) GetNode(uuid string) (*Node, bool) {
|
||||||
t.mu.RLock()
|
t.mu.RLock()
|
||||||
@@ -300,4 +364,3 @@ func (t *Tracker) GetOfflineServers() []string {
|
|||||||
}
|
}
|
||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -14,8 +14,8 @@ import (
|
|||||||
"nukumizu-backend/database"
|
"nukumizu-backend/database"
|
||||||
"nukumizu-backend/internal/controller"
|
"nukumizu-backend/internal/controller"
|
||||||
"nukumizu-backend/internal/controller/pipes"
|
"nukumizu-backend/internal/controller/pipes"
|
||||||
"nukumizu-backend/internal/komari"
|
|
||||||
"nukumizu-backend/internal/controller/pipes/qq_napcat"
|
"nukumizu-backend/internal/controller/pipes/qq_napcat"
|
||||||
|
"nukumizu-backend/internal/komari"
|
||||||
"nukumizu-backend/internal/node"
|
"nukumizu-backend/internal/node"
|
||||||
"nukumizu-backend/postLog"
|
"nukumizu-backend/postLog"
|
||||||
"nukumizu-backend/utils"
|
"nukumizu-backend/utils"
|
||||||
@@ -289,4 +289,19 @@ func startBackgroundTasks(_ *config.Config) {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Safety net for the initial refresh gate: if some node never reports a
|
||||||
|
// status, force-complete the initial refresh after a grace period so status
|
||||||
|
// notifications cannot stay blocked forever.
|
||||||
|
go func() {
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
postLog.Error(fmt.Sprintf("Bootstrap timeout panic: %v", r))
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
time.Sleep(2 * time.Minute) // Add 2 minutes grace period to allow nodes to report status.
|
||||||
|
if tracker := node.GetTracker(); tracker != nil {
|
||||||
|
tracker.CompleteBootstrap()
|
||||||
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user