{"record":{"id":"7d60ffe796d210c5","repo":"risingwavelabs/risingwave","slug":"graph-is-not-a-dag","errorCode":null,"errorMessage":"graph is not a DAG","messagePattern":"graph is not a DAG","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":2149,"sourceCode":"        }\n\n        let mut i = 0;\n        while let Some(&fragment_id) = topo.get(i) {\n            i += 1;\n            // Find if we can process more fragments.\n            for (upstream_job_id, _) in self.get_upstreams(fragment_id) {\n                let downstream_cnt = downstream_cnts.get_mut(&upstream_job_id).unwrap();\n                *downstream_cnt -= 1;\n                if *downstream_cnt == 0 {\n                    downstream_cnts.remove(&upstream_job_id);\n                    topo.push(upstream_job_id);\n                }\n            }\n        }\n\n        if !downstream_cnts.is_empty() {\n            // There are fragments that are not processed yet.\n            bail!(\"graph is not a DAG\");\n        }\n\n        Ok(topo)\n    }\n\n    /// Seal a [`BuildingFragment`] from the graph into a [`Fragment`], which will be further used\n    /// to build actors on the compute nodes and persist into meta store.\n    pub(super) fn seal_fragment(\n        &self,\n        id: GlobalFragmentId,\n        distribution: Distribution,\n        stream_node: StreamNode,\n    ) -> Fragment {\n        let building_fragment = self.get_fragment(id).into_building().unwrap();\n        let internal_tables = building_fragment.extract_internal_tables();\n        let BuildingFragment {\n            inner,\n            job_id,","sourceCodeStart":2131,"sourceCodeEnd":2167,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L2131-L2167","documentation":"During topological sorting of the stream fragment graph, all fragments should be consumed (their downstream counts reduced to zero). If `downstream_cnts` still has entries after the traversal, the fragment graph contains a cycle (or unreachable nodes), so the builder bails with 'graph is not a DAG'. A cyclic fragment graph cannot be scheduled.","triggerScenarios":"Building a stream graph whose fragment dependency graph contains a cycle — e.g. fragments referencing each other circularly after building edges — so Kahn's algorithm finishes with unprocessed fragments remaining in `downstream_cnts`.","commonSituations":"Internal meta bugs in edge construction creating circular dependencies between fragments; corrupted job/fragment metadata from a failed migration or manual edits to the catalog.","solutions":["Inspect the fragment graph for the job (dump fragment dependencies) and find the cycle; fix the edge-building code that introduced it.","Recreate the streaming job so a fresh, acyclic graph is built.","If metadata corruption is suspected, drop and rebuild the job from source DDL rather than repairing fragments."],"exampleFix":"null","handlingStrategy":"validation","validationCode":"// Verify acyclicity of the fragment dependency graph before topo-sort\nfn is_acyclic(nodes: &[FragmentId], edges: &[(FragmentId, FragmentId)]) -> bool {\n    // Kahn's algorithm: if processed < nodes.len(), a cycle exists\n    let mut indeg: HashMap<_, usize> = nodes.iter().map(|n| (*n, 0)).collect();\n    let mut adj: HashMap<_, Vec<_>> = HashMap::new();\n    for (u, v) in edges { *indeg.entry(*v).or_default() += 1; adj.entry(*u).or_default().push(*v); }\n    let mut q: Vec<_> = nodes.iter().filter(|n| indeg[n] == 0).map(|n| *n).collect();\n    let mut done = 0;\n    while let Some(u) = q.pop() { done += 1; for v in adj.get(&u).cloned().unwrap_or_default() { let d = indeg.get_mut(&v).unwrap(); *d -= 1; if *d == 0 { q.push(v); } } }\n    done == nodes.len()\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Add an acyclicity unit test to edge construction in the stream graph builder","Avoid manual edits/migrations that introduce circular fragment references","Recreate jobs whose fragment metadata may be corrupted instead of repairing"],"tags":["meta","streaming","graph","cycle"],"backgroundTag":"internal-invariant-violation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}