Discord 语音全平台宕机：一个隐藏的循环依赖如何引发级联崩溃

信令 → 状态 → 配置 → 信令

"""
service_cycle_detector.py
检测微服务依赖图中的所有循环依赖。

使用方法：
  python service_cycle_detector.py

把 SERVICES 字典替换成你自己的服务调用关系即可。
"""

from collections import defaultdict

# ---- 替换成你的服务依赖关系 ----
# key: 服务名, value: 该服务直接依赖（调用）的其他服务列表
SERVICES = {
    "signaling":   ["state", "media"],
    "state":       ["config", "db"],
    "config":      ["signaling", "db"],   # config → signaling 构成环路
    "media":       ["state", "config"],
    "db":          [],                     # 基础服务，无下游依赖
    "gateway":     ["signaling", "state"],
}


def find_cycles(graph: dict[str, list[str]]) -> list[list[str]]:
    """用 DFS 检测有向图中的所有环，返回每个环的节点路径列表。"""
    visited = set()
    in_stack = set()
    cycles = []

    def dfs(node: str, path: list[str]):
        visited.add(node)
        in_stack.add(node)
        path.append(node)

        for neighbor in graph.get(node, []):
            if neighbor not in visited:
                dfs(neighbor, path)
            elif neighbor in in_stack:
                # 找到环：从 path 中截取环的部分
                cycle_start = path.index(neighbor)
                cycle = path[cycle_start:] + [neighbor]
                cycles.append(cycle)

        path.pop()
        in_stack.discard(node)

    for node in graph:
        if node not in visited:
            dfs(node, [])

    return cycles


def main():
    cycles = find_cycles(SERVICES)
    if not cycles:
        print("✅ 未检测到循环依赖，依赖图是安全的。")
    else:
        print(f"⚠️  检测到 {len(cycles)} 个循环依赖：")
        for i, cycle in enumerate(cycles, 1):
            chain = " → ".join(cycle)
            print(f"  环 {i}: {chain}")
            # 标注隐式风险
            involved = set(cycle[:-1])
            print(f"       涉及服务: {', '.join(involved)}")
            print(f"       环长度: {len(cycle) - 1}")
            print()

    # 额外输出：每个服务的依赖深度（最长调用链）
    print("--- 依赖深度分析 ---")
    depth_cache = {}

    def max_depth(node: str) -> int:
        if node in depth_cache:
            return depth_cache[node]
        deps = graph.get(node, [])
        if not deps:
            depth_cache[node] = 0
            return 0
        # 如果存在环，深度可能无限；这里设上限 10 防止递归溢出
        depth_cache[node] = -1  # 标记为"正在计算"
        d = 1 + max((max_depth(d) for d in deps), default=0)
        if depth_cache[node] == -1:
            depth_cache[node] = min(d, 10)
        return depth_cache[node]

    graph = SERVICES
    for svc in sorted(graph):
        d = max_depth(svc)
        flag = "🔴" if d >= 5 else ("🟡" if d >= 3 else "🟢")
        print(f"  {flag} {svc}: 最大依赖深度 {d}")


if __name__ == "__main__":
    main()

⚠️  检测到 1 个循环依赖：
  环 1: signaling → state → config → signaling
       涉及服务: signaling, state, config
       环长度: 3

--- 依赖深度分析 ---
  🟢 db: 最大依赖深度 0
  🟡 config: 最大依赖深度 3
  🟡 gateway: 最大依赖深度 4
  🟡 media: 最大依赖深度 3
  🟡 signaling: 最大依赖深度 3
  🟡 state: 最大依赖深度 2