From b05219ef4cc8dd0d77d0a30b4d043d5f66eeacd3 Mon Sep 17 00:00:00 2001 From: douxu Date: Tue, 7 Jul 2026 10:01:14 +0800 Subject: [PATCH] refactor: replace topology tree with graph cache - replace globalTree startup loading with globalTopologyGraph - add topologyGraph adjacency cache with in/out edges, start nodes, and end nodes - remove multiBranchTreeNode and legacy tree-building helpers - make topology queries independent of the all-zero UUID virtual root - keep TOPOLOGY_ANALYSIS on per-task DB querying while clarifying point-to-point reachability results - add coverage for multi-parent topology reachability --- database/query_topologic.go | 93 ++-------------------- diagram/multi_branch_tree.go | 125 ----------------------------- diagram/topology_graph.go | 138 +++++++++++++++++++++++++++++++++ diagram/topology_graph_test.go | 35 +++++++++ main.go | 4 +- sql/topologic.go | 15 ++++ task/handler_factory.go | 28 +++++-- 7 files changed, 219 insertions(+), 219 deletions(-) delete mode 100644 diagram/multi_branch_tree.go create mode 100644 diagram/topology_graph.go create mode 100644 diagram/topology_graph_test.go diff --git a/database/query_topologic.go b/database/query_topologic.go index 4799875..3d613eb 100644 --- a/database/query_topologic.go +++ b/database/query_topologic.go @@ -3,11 +3,8 @@ package database import ( "context" - "fmt" "time" - "modelRT/constants" - "modelRT/diagram" "modelRT/logger" "modelRT/orm" "modelRT/sql" @@ -24,16 +21,19 @@ func QueryTopologic(ctx context.Context, tx *gorm.DB) ([]orm.Topologic, error) { cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() - result := tx.WithContext(cancelCtx).Clauses(clause.Locking{Strength: "UPDATE"}).Raw(sql.RecursiveSQL, constants.UUIDNilStr).Scan(&topologics) + result := tx.WithContext(cancelCtx). + Clauses(clause.Locking{Strength: "UPDATE"}). + Find(&topologics) if result.Error != nil { - logger.Error(ctx, "query circuit diagram topologic info by start node uuid failed", "start_node_uuid", constants.UUIDNilStr, "error", result.Error) + logger.Error(ctx, "query circuit diagram topologic info failed", "error", result.Error) return nil, result.Error } return topologics, nil } -// QueryTopologicByStartUUID returns all edges reachable from startUUID following -// directed uuid_from → uuid_to edges in the topologic table. +// QueryTopologicByStartUUID returns all directed edges reachable from startUUID. +// It is used by point-to-point topology reachability checks and intentionally +// does not depend on the legacy all-zero UUID virtual root. func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.UUID) ([]orm.Topologic, error) { var topologics []orm.Topologic @@ -42,7 +42,7 @@ func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid. result := tx.WithContext(cancelCtx). Clauses(clause.Locking{Strength: "UPDATE"}). - Raw(sql.RecursiveSQL, startUUID). + Raw(sql.RecursiveTopologicByStartSQL, startUUID). Scan(&topologics) if result.Error != nil { logger.Error(ctx, "query topologic by start uuid failed", "start_uuid", startUUID, "error", result.Error) @@ -50,80 +50,3 @@ func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid. } return topologics, nil } - -// QueryTopologicFromDB return the result of query topologic info from DB. -// Returns the root node and a flat nodeMap for O(1) lookup by UUID. -func QueryTopologicFromDB(ctx context.Context, tx *gorm.DB) (*diagram.MultiBranchTreeNode, map[uuid.UUID]*diagram.MultiBranchTreeNode, error) { - topologicInfos, err := QueryTopologic(ctx, tx) - if err != nil { - logger.Error(ctx, "query topologic info failed", "error", err) - return nil, nil, err - } - - tree, nodeMap, err := BuildMultiBranchTree(topologicInfos) - if err != nil { - logger.Error(ctx, "init topologic failed", "error", err) - return nil, nil, err - } - return tree, nodeMap, nil -} - -// BuildMultiBranchTree return the multi branch tree by topologic info. -// Returns the root node and a flat nodeMap for O(1) lookup by UUID. -func BuildMultiBranchTree(topologics []orm.Topologic) (*diagram.MultiBranchTreeNode, map[uuid.UUID]*diagram.MultiBranchTreeNode, error) { - nodeMap := make(map[uuid.UUID]*diagram.MultiBranchTreeNode, len(topologics)*2) - - for _, topo := range topologics { - if _, exists := nodeMap[topo.UUIDFrom]; !exists { - // UUIDNil is the virtual root sentinel — skip creating a regular node for it - if topo.UUIDFrom != constants.UUIDNil { - nodeMap[topo.UUIDFrom] = &diagram.MultiBranchTreeNode{ - ID: topo.UUIDFrom, - Children: make([]*diagram.MultiBranchTreeNode, 0), - } - } - } - - if _, exists := nodeMap[topo.UUIDTo]; !exists { - if topo.UUIDTo != constants.UUIDNil { - nodeMap[topo.UUIDTo] = &diagram.MultiBranchTreeNode{ - ID: topo.UUIDTo, - Children: make([]*diagram.MultiBranchTreeNode, 0), - } - } - } - } - - for _, topo := range topologics { - var parent *diagram.MultiBranchTreeNode - if topo.UUIDFrom == constants.UUIDNil { - if _, exists := nodeMap[constants.UUIDNil]; !exists { - nodeMap[constants.UUIDNil] = &diagram.MultiBranchTreeNode{ - ID: constants.UUIDNil, - Children: make([]*diagram.MultiBranchTreeNode, 0), - } - } - parent = nodeMap[constants.UUIDNil] - } else { - parent = nodeMap[topo.UUIDFrom] - } - - var child *diagram.MultiBranchTreeNode - if topo.UUIDTo == constants.UUIDNil { - child = &diagram.MultiBranchTreeNode{ - ID: topo.UUIDTo, - } - } else { - child = nodeMap[topo.UUIDTo] - } - child.Parent = parent - parent.Children = append(parent.Children, child) - } - - // return root vertex - root, exists := nodeMap[constants.UUIDNil] - if !exists { - return nil, nil, fmt.Errorf("root node not found") - } - return root, nodeMap, nil -} diff --git a/diagram/multi_branch_tree.go b/diagram/multi_branch_tree.go deleted file mode 100644 index d61e385..0000000 --- a/diagram/multi_branch_tree.go +++ /dev/null @@ -1,125 +0,0 @@ -// Package diagram provide diagram data structure and operation -package diagram - -import ( - "fmt" - - "github.com/gofrs/uuid" -) - -var GlobalTree *MultiBranchTreeNode - -// MultiBranchTreeNode represents a topological structure using an multi branch tree -type MultiBranchTreeNode struct { - ID uuid.UUID // 节点唯一标识 - Parent *MultiBranchTreeNode // 指向父节点的指针 - Children []*MultiBranchTreeNode // 指向所有子节点的指针切片 -} - -func NewMultiBranchTree(id uuid.UUID) *MultiBranchTreeNode { - return &MultiBranchTreeNode{ - ID: id, - Children: make([]*MultiBranchTreeNode, 0), - } -} - -func (n *MultiBranchTreeNode) AddChild(child *MultiBranchTreeNode) { - child.Parent = n - n.Children = append(n.Children, child) -} - -func (n *MultiBranchTreeNode) RemoveChild(childID uuid.UUID) bool { - for i, child := range n.Children { - if child.ID == childID { - n.Children = append(n.Children[:i], n.Children[i+1:]...) - child.Parent = nil - return true - } - } - return false -} - -func (n *MultiBranchTreeNode) FindNodeByID(id uuid.UUID) *MultiBranchTreeNode { - if n.ID == id { - return n - } - - for _, child := range n.Children { - if found := child.FindNodeByID(id); found != nil { - return found - } - } - return nil -} - -func (n *MultiBranchTreeNode) PrintTree(level int) { - for range level { - fmt.Print(" ") - } - - fmt.Printf("-ID: %s\n", n.ID) - - for _, child := range n.Children { - child.PrintTree(level + 1) - } -} - -// FindPath returns the ordered node sequence from startID to endID using the -// supplied nodeMap for O(1) lookup. It walks each node up to the root to find -// the LCA, then stitches the two half-paths together. -// Returns nil when either node is absent from nodeMap or no path exists. -func FindPath(startID, endID uuid.UUID, nodeMap map[uuid.UUID]*MultiBranchTreeNode) []*MultiBranchTreeNode { - startNode, ok := nodeMap[startID] - if !ok { - return nil - } - endNode, ok := nodeMap[endID] - if !ok { - return nil - } - - // collect ancestors (inclusive) from a node up to the root sentinel - ancestors := func(n *MultiBranchTreeNode) []*MultiBranchTreeNode { - var chain []*MultiBranchTreeNode - for n != nil { - chain = append(chain, n) - n = n.Parent - } - return chain - } - - startChain := ancestors(startNode) // [start, ..., root] - endChain := ancestors(endNode) // [end, ..., root] - - // index startChain by ID for fast LCA detection - startIdx := make(map[uuid.UUID]int, len(startChain)) - for i, node := range startChain { - startIdx[node.ID] = i - } - - // find LCA: first node in endChain that also appears in startChain - lcaEndPos := -1 - lcaStartPos := -1 - for i, node := range endChain { - if j, found := startIdx[node.ID]; found { - lcaEndPos = i - lcaStartPos = j - break - } - } - - if lcaEndPos < 0 { - return nil // disconnected - } - - // path = startChain[0..lcaStartPos] reversed + endChain[lcaEndPos..0] reversed - path := make([]*MultiBranchTreeNode, 0, lcaStartPos+lcaEndPos+1) - for i := 0; i <= lcaStartPos; i++ { - path = append(path, startChain[i]) - } - // append end-side (skip LCA to avoid duplication), reversed - for i := lcaEndPos - 1; i >= 0; i-- { - path = append(path, endChain[i]) - } - return path -} diff --git a/diagram/topology_graph.go b/diagram/topology_graph.go new file mode 100644 index 0000000..636cbfa --- /dev/null +++ b/diagram/topology_graph.go @@ -0,0 +1,138 @@ +package diagram + +import ( + "sync" + + "modelRT/orm" + + "github.com/gofrs/uuid" +) + +// TopologyGraph represents directed topologic links with adjacency lists. +// It preserves multiple parents for one node. +type TopologyGraph struct { + Nodes map[uuid.UUID]struct{} + OutEdges map[uuid.UUID][]uuid.UUID + InEdges map[uuid.UUID][]uuid.UUID + StartNodes []uuid.UUID + EndNodes []uuid.UUID +} + +var ( + globalTopologyGraphMu sync.RWMutex + GlobalTopologyGraph *TopologyGraph +) + +// NewTopologyGraph builds a directed graph cache from topologic edges. +func NewTopologyGraph(edges []orm.Topologic) *TopologyGraph { + graph := &TopologyGraph{ + Nodes: make(map[uuid.UUID]struct{}, len(edges)*2), + OutEdges: make(map[uuid.UUID][]uuid.UUID, len(edges)), + InEdges: make(map[uuid.UUID][]uuid.UUID, len(edges)), + } + + for _, edge := range edges { + from := edge.UUIDFrom + to := edge.UUIDTo + + graph.Nodes[from] = struct{}{} + graph.Nodes[to] = struct{}{} + graph.OutEdges[from] = append(graph.OutEdges[from], to) + graph.InEdges[to] = append(graph.InEdges[to], from) + } + + graph.StartNodes = graph.findStartNodes() + graph.EndNodes = graph.findEndNodes() + return graph +} + +// SetGlobalTopologyGraph replaces the process-wide topology graph cache. +func SetGlobalTopologyGraph(graph *TopologyGraph) { + globalTopologyGraphMu.Lock() + defer globalTopologyGraphMu.Unlock() + GlobalTopologyGraph = graph +} + +// GetGlobalTopologyGraph returns the process-wide topology graph cache. +func GetGlobalTopologyGraph() *TopologyGraph { + globalTopologyGraphMu.RLock() + defer globalTopologyGraphMu.RUnlock() + return GlobalTopologyGraph +} + +func (g *TopologyGraph) findStartNodes() []uuid.UUID { + startNodes := make([]uuid.UUID, 0) + for id := range g.Nodes { + if len(g.InEdges[id]) == 0 && len(g.OutEdges[id]) > 0 { + startNodes = append(startNodes, id) + } + } + return startNodes +} + +func (g *TopologyGraph) findEndNodes() []uuid.UUID { + endNodes := make([]uuid.UUID, 0) + for id := range g.Nodes { + if len(g.InEdges[id]) > 0 && len(g.OutEdges[id]) == 0 { + endNodes = append(endNodes, id) + } + } + return endNodes +} + +// IsReachable reports whether end can be reached from start following directed +// uuid_from -> uuid_to edges. +func (g *TopologyGraph) IsReachable(start, end uuid.UUID) bool { + return len(g.FindPath(start, end)) > 0 +} + +// FindPath returns one shortest directed path from start to end, or nil when +// no directed path exists. +func (g *TopologyGraph) FindPath(start, end uuid.UUID) []uuid.UUID { + if g == nil { + return nil + } + if start == end { + if _, exists := g.Nodes[start]; exists { + return []uuid.UUID{start} + } + return nil + } + + visited := map[uuid.UUID]struct{}{start: {}} + parent := make(map[uuid.UUID]uuid.UUID) + queue := []uuid.UUID{start} + + for len(queue) > 0 { + cur := queue[0] + queue = queue[1:] + + for _, next := range g.OutEdges[cur] { + if _, seen := visited[next]; seen { + continue + } + visited[next] = struct{}{} + parent[next] = cur + + if next == end { + return reconstructTopologyGraphPath(parent, start, end) + } + + queue = append(queue, next) + } + } + + return nil +} + +func reconstructTopologyGraphPath(parent map[uuid.UUID]uuid.UUID, start, end uuid.UUID) []uuid.UUID { + path := make([]uuid.UUID, 0) + for cur := end; cur != start; cur = parent[cur] { + path = append(path, cur) + } + path = append(path, start) + for i, j := 0, len(path)-1; i < j; i, j = i+1, j-1 { + path[i], path[j] = path[j], path[i] + } + return path +} diff --git a/diagram/topology_graph_test.go b/diagram/topology_graph_test.go new file mode 100644 index 0000000..556fdbf --- /dev/null +++ b/diagram/topology_graph_test.go @@ -0,0 +1,35 @@ +package diagram + +import ( + "testing" + + "modelRT/orm" + + "github.com/gofrs/uuid" +) + +func TestTopologyGraphSupportsMultiParentReachability(t *testing.T) { + startA := uuid.Must(uuid.NewV4()) + startB := uuid.Must(uuid.NewV4()) + shared := uuid.Must(uuid.NewV4()) + end := uuid.Must(uuid.NewV4()) + + graph := NewTopologyGraph([]orm.Topologic{ + {UUIDFrom: startA, UUIDTo: shared}, + {UUIDFrom: startB, UUIDTo: shared}, + {UUIDFrom: shared, UUIDTo: end}, + }) + + if len(graph.StartNodes) != 2 { + t.Fatalf("expected 2 start nodes, got %d", len(graph.StartNodes)) + } + if len(graph.InEdges[shared]) != 2 { + t.Fatalf("expected shared node to keep 2 parents, got %d", len(graph.InEdges[shared])) + } + if !graph.IsReachable(startA, end) { + t.Fatalf("expected %s to reach %s", startA, end) + } + if !graph.IsReachable(startB, end) { + t.Fatalf("expected %s to reach %s", startB, end) + } +} diff --git a/main.go b/main.go index 84fa153..8701c63 100644 --- a/main.go +++ b/main.go @@ -242,12 +242,12 @@ func main() { } go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement) - tree, _, err := database.QueryTopologicFromDB(ctx, tx) + topologics, err := database.QueryTopologic(ctx, tx) if err != nil { logger.Error(ctx, "load topologic info from postgres failed", "error", err) panic(err) } - diagram.GlobalTree = tree + diagram.SetGlobalTopologyGraph(diagram.NewTopologyGraph(topologics)) return nil }) diff --git a/sql/topologic.go b/sql/topologic.go index 07d1733..f440e27 100644 --- a/sql/topologic.go +++ b/sql/topologic.go @@ -12,3 +12,18 @@ var RecursiveSQL = `WITH RECURSIVE recursive_tree as ( JOIN recursive_tree rt ON t.uuid_from = rt.uuid_to ) SELECT * FROM recursive_tree;` + +// RecursiveTopologicByStartSQL returns every directed edge reachable from the +// supplied start component. It tracks the visited node path inside PostgreSQL +// so cycles in topologic data cannot recurse forever. +var RecursiveTopologicByStartSQL = `WITH RECURSIVE recursive_tree as ( + SELECT uuid_from, uuid_to, flag, ARRAY[uuid_from, uuid_to] AS path + FROM "topologic" + WHERE uuid_from = ? + UNION ALL + SELECT t.uuid_from, t.uuid_to, t.flag, rt.path || t.uuid_to + FROM "topologic" t + JOIN recursive_tree rt ON t.uuid_from = rt.uuid_to + WHERE NOT t.uuid_to = ANY(rt.path) +) +SELECT uuid_from, uuid_to, flag FROM recursive_tree;` diff --git a/task/handler_factory.go b/task/handler_factory.go index 26e295a..af88a65 100644 --- a/task/handler_factory.go +++ b/task/handler_factory.go @@ -98,10 +98,10 @@ func NewTopologyAnalysisHandler() *TopologyAnalysisHandler { } } -// Execute processes a topology analysis task. +// Execute processes a point-to-point topology reachability task. // Params (all sourced from the MQ message, no DB lookup needed): -// - start_component_uuid (string, required): BFS origin -// - end_component_uuid (string, required): reachability target +// - start_component_uuid (string, required): directed traversal origin +// - end_component_uuid (string, required): directed reachability target // - check_in_service (bool, optional, default true): skip out-of-service components func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID, params map[string]any, db *gorm.DB) error { logger.Info(ctx, "topology analysis started", "task_id", taskID) @@ -123,7 +123,8 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID, logger.Warn(ctx, "update progress failed", "task_id", taskID, "progress", 20, "error", err) } - // Phase 2: query topology edges from startComponentUUID, build adjacency list + // Phase 2: query only edges reachable from startComponentUUID, then build + // the adjacency list used for point-to-point directed reachability. topoEdges, err := database.QueryTopologicByStartUUID(ctx, db, startComponentUUID) if err != nil { return fmt.Errorf("query topology from start node: %w", err) @@ -159,7 +160,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID, // check the start node itself before BFS if !inServiceMap[startComponentUUID] { return persistTopologyResult(ctx, db, taskID, startComponentUUID, endComponentUUID, - checkInService, false, nil, &startComponentUUID) + checkInService, false, nil, &startComponentUUID, 0) } } @@ -167,7 +168,8 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID, logger.Warn(ctx, "update progress failed", "task_id", taskID, "progress", 60, "error", err) } - // Phase 4: BFS reachability check + // Phase 4: point-to-point BFS reachability check. Multiple parents and + // multiple paths to a node are valid; visited only prevents cycles/rework. visited := make(map[uuid.UUID]struct{}) parent := make(map[uuid.UUID]uuid.UUID) // for path reconstruction queue := []uuid.UUID{startComponentUUID} @@ -214,7 +216,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID, } return persistTopologyResult(ctx, db, taskID, startComponentUUID, endComponentUUID, - checkInService, isReachable, path, blockedBy) + checkInService, isReachable, path, blockedBy, len(visited)) } // parseTopologyAnalysisParams extracts and validates the three required fields. @@ -270,6 +272,7 @@ func persistTopologyResult( ctx context.Context, db *gorm.DB, taskID uuid.UUID, startID, endID uuid.UUID, checkInService, isReachable bool, path []uuid.UUID, blockedBy *uuid.UUID, + visitedCount int, ) error { pathStrs := make([]string, 0, len(path)) for _, id := range path { @@ -281,11 +284,22 @@ func persistTopologyResult( "end_component_uuid": endID.String(), "check_in_service": checkInService, "is_reachable": isReachable, + "analysis_type": "POINT_TO_POINT_REACHABILITY", "path": pathStrs, + "path_node_count": len(pathStrs), + "visited_count": visitedCount, "computed_at": time.Now().Unix(), } + if isReachable { + result["hop_count"] = len(pathStrs) - 1 + } if blockedBy != nil { result["blocked_by"] = blockedBy.String() + result["reason"] = "OUT_OF_SERVICE_COMPONENT" + } else if isReachable { + result["reason"] = "REACHABLE" + } else { + result["reason"] = "NO_DIRECTED_PATH" } if err := database.CreateAsyncTaskResult(ctx, db, taskID, result); err != nil {