From 981000667a8e86f8d98a5924135042bfd11940f1 Mon Sep 17 00:00:00 2001 From: Sahil Lenka Date: Sat, 3 Oct 2026 04:59:24 +0530 Subject: [PATCH] fix: reject file edge labels that have more than one source node An edge label in mkconcore is a single volume with one writer. edges_dict keeps the source of the first edge with a given label and silently ignores the source of any later edge with the same label, so the second source node never gets an out mount and everything it writes on that edge is lost. The target also ends up mounting the same volume twice (in1 and in2). mkconcore now stops with an error when a file edge label has more than one source node, and concore validate reports the same case so it gets caught before building. --- concore_cli/commands/validate.py | 17 +++++++++++++++- mkconcore.py | 11 ++++++++++- tests/test_cli.py | 33 +++++++++++++++++++++++++++++++ tests/test_graph.py | 34 ++++++++++++++++++++++++++++++++ 4 files changed, 93 insertions(+), 2 deletions(-) diff --git a/concore_cli/commands/validate.py b/concore_cli/commands/validate.py index a7168b3..08d9f66 100644 --- a/concore_cli/commands/validate.py +++ b/concore_cli/commands/validate.py @@ -64,6 +64,8 @@ def _classify_message(message, bucket_name): return {"error_type": "missing_edge_source"} if message.startswith("Edge references non-existent target node:"): return {"error_type": "missing_edge_target"} + if message.startswith("Edge label '") and "has more than one source" in message: + return {"error_type": "edge_label_multiple_sources"} if message == "Workflow contains cycles (expected for control loops)": return {"error_type": "cycle_detected"} if message.startswith("Invalid port number:"): @@ -292,6 +294,7 @@ def finalize(): edge_label_regex = re.compile(r"0x([a-fA-F0-9]+)_(\S+)") zmq_edges = 0 file_edges = 0 + label_sources = {} for edge in edges: try: @@ -300,13 +303,25 @@ def finalize(): label_tag = edge.find("EdgeLabel") if label_tag and label_tag.text: - if edge_label_regex.match(label_tag.text.strip()): + edge_label = label_tag.text.strip() + if edge_label_regex.match(edge_label): zmq_edges += 1 else: file_edges += 1 + label_sources.setdefault(edge_label, set()).add( + edge.get("source") + ) except Exception: pass + # mkconcore mounts a file edge for its first source only + for edge_label, sources in label_sources.items(): + if len(sources) > 1: + errors.append( + f"Edge label '{edge_label}' has more than one source node: " + f"{', '.join(sorted(sources))}" + ) + if zmq_edges > 0: info.append(f"ZMQ-based edges: {zmq_edges}") if file_edges > 0: diff --git a/mkconcore.py b/mkconcore.py index 987eadf..7eccfbe 100644 --- a/mkconcore.py +++ b/mkconcore.py @@ -382,8 +382,17 @@ def cleanup_script_files(): #Validate edge labels safe_name(edge_label, f"Edge label '{edge_label}'") + source_label = nodes_dict[edge['source']] if edge_label not in edges_dict: - edges_dict[edge_label] = [nodes_dict[edge['source']], []] + edges_dict[edge_label] = [source_label, []] + elif edges_dict[edge_label][0] != source_label: + # an edge label is one volume with a single writer, a second + # source would get no out mount and its writes would be lost + logging.error( + f"Edge label '{raw_label}' has more than one source node: " + f"{edges_dict[edge_label][0]} and {source_label}" + ) + sys.exit(1) edges_dict[edge_label][1].append(nodes_dict[edge['target']]) except (IndexError, AttributeError, KeyError): logging.debug('An edge with no valid properties or missing node encountered and ignored') diff --git a/tests/test_cli.py b/tests/test_cli.py index d746040..8adff76 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -136,6 +136,39 @@ def test_build_command_missing_source(self): ) self.assertNotEqual(result.exit_code, 0) + def test_build_command_rejects_edge_label_with_multiple_sources(self): + with self.runner.isolated_filesystem(temp_dir=self.temp_dir): + Path("src").mkdir() + for name in ["a.py", "b.py", "c.py"]: + Path("src", name).write_text("import concore\n") + Path("workflow.graphml").write_text( + '\n' + '\n' + 'A:a.py\n' + 'B:b.py\n' + 'C:c.py\n' + 'shared\n' + 'shared\n' + "\n" + "\n" + ) + + result = self.runner.invoke( + cli, + [ + "build", + "workflow.graphml", + "--source", + "src", + "--output", + "out", + "--type", + "posix", + ], + ) + self.assertNotEqual(result.exit_code, 0) + self.assertIn("Edge label 'shared' has more than one source", result.output) + def test_build_command_from_project_dir(self): with self.runner.isolated_filesystem(temp_dir=self.temp_dir): result = self.runner.invoke(cli, ["init", "test-project"]) diff --git a/tests/test_graph.py b/tests/test_graph.py index efebe3e..b26859c 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -1,3 +1,4 @@ +import json import unittest import tempfile import shutil @@ -188,6 +189,39 @@ def test_validate_with_existing_source_file(self): self.assertIn("Validation passed", result.output) + def test_validate_edge_label_with_multiple_sources(self): + content = """ + + + + A:a.py + + + B:b.py + + + C:c.py + + + shared + + + shared + + + + """ + filepath = self.create_graph_file("shared_label.graphml", content) + + result = self.runner.invoke(cli, ["validate", filepath]) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("Edge label 'shared' has more than one source", result.output) + + result = self.runner.invoke(cli, ["validate", filepath, "--format", "json"]) + error_types = [e["error_type"] for e in json.loads(result.output)["errors"]] + self.assertIn("edge_label_multiple_sources", error_types) + def test_validate_zmq_port_conflict(self): content = """