From 4a3778386da5fefc1faf95eb3ad0abd2849aad3c Mon Sep 17 00:00:00 2001 From: Raghuvarma905 Date: Tue, 20 Oct 2026 00:34:49 +0530 Subject: [PATCH 1/4] MERGE between bound nodes keys a relationship on a value from the row UNWIND $rows AS r MATCH (s:Account {id: r.src}), (d:Account {id: r.dst}) MERGE (s)-[t:TRANSFER {id: r.id}]->(d) is the bulk edge-upsert idiom, and it lost data without an error. The planner hands a MERGE between already-bound endpoints to MatchMergeEdgeOperator, and it handed over only the literal half of the relationship's property map; `{id: r.id}` lives in `property_exprs`, which was dropped. So the operator matched any :TRANSFER between the pair and, when it created one, created it without `id`. Two rows for one pair wrote one edge, and the statement reported both as written. #642 fixed the same defect for nodes and for whole-pattern MERGE; this is the relationship-between-bound-nodes half it left. Found loading the INT-10 reference pipeline (#1715): 60 transfers became 55 edges. The fix resolves the expressions against each row with the resolver node MERGE already uses, and uses the one resolved map both to match and to create -- a MERGE that searched on one set of values and wrote another would make a second edge on every re-run. EXPLAIN now shows the expressions too. tests/merge_row_properties.rs gains six cases: distinct keys between one pair, an existing edge with a different key left alone, a key from another bound node, literal and row properties together, ON CREATE / ON MATCH, and an undirected pattern. All six fail without the change and pass with it; the five existing cases pass either way. The openCypher TCK reads the same as main (4 failures, the same 4). --- src/query/executor/operator.rs | 41 +++++++++++++-- src/query/executor/planner.rs | 6 ++- tests/merge_row_properties.rs | 95 ++++++++++++++++++++++++++++++++++ 3 files changed, 137 insertions(+), 5 deletions(-) diff --git a/src/query/executor/operator.rs b/src/query/executor/operator.rs index 5d43f6e0..eac9b53f 100644 --- a/src/query/executor/operator.rs +++ b/src/query/executor/operator.rs @@ -16962,6 +16962,14 @@ pub struct MatchMergeEdgeOperator { /// `source -> target` and MERGE created a duplicate beside an existing /// `target -> source` (#938). edges_to_merge: Vec<(String, String, EdgeType, HashMap, Option, bool)>, + /// The non-literal half of each relationship's property map, by position + /// in `edges_to_merge`: `{id: r.id}` from an UNWIND, `{k: a.name}` from a + /// bound node. `edges_to_merge` carries only the literals, and these were + /// never handed over, so `UNWIND $rows AS r MATCH (a), (b) MERGE + /// (a)-[:T {id: r.id}]->(b)` matched any `:T` between the pair and created + /// edges without the property -- two rows for one pair wrote one edge, + /// with no `id`, and reported success. #642 fixed the node half. + property_exprs: Vec>>, on_create_set: Vec<(String, String, Expression)>, on_match_set: Vec<(String, String, Expression)>, /// Whole-entity `ON CREATE`/`ON MATCH SET`; see `MergeOperator` (#874). @@ -16982,6 +16990,7 @@ impl MatchMergeEdgeOperator { Self { input, edges_to_merge, + property_exprs: Vec::new(), on_create_set, on_match_set, on_create_entity_set: Vec::new(), @@ -17002,6 +17011,13 @@ impl MatchMergeEdgeOperator { self.on_match_entity_set = on_match; self } + + /// Attach each relationship's non-literal properties, in the order of + /// `edges_to_merge`. A missing entry means the pattern has none. + pub fn with_property_exprs(mut self, exprs: Vec>>) -> Self { + self.property_exprs = exprs; + self + } } impl PhysicalOperator for MatchMergeEdgeOperator { @@ -17016,9 +17032,22 @@ impl PhysicalOperator for MatchMergeEdgeOperator { fn next_mut(&mut self, store: &mut GraphStore, tenant_id: &str) -> ExecutionResult> { if !self.done { while let Some(record) = self.input.next_mut(store, tenant_id)? { - for (source_var, target_var, edge_type, properties, edge_var, undirected) in - &self.edges_to_merge + for (i, (source_var, target_var, edge_type, literal_properties, edge_var, undirected)) in + self.edges_to_merge.iter().enumerate() { + // Resolved against this row, and the one map used both to + // match and to create -- the property #642 holds to for + // nodes: a MERGE that searched on one set of values and + // wrote another would create an edge its own pattern + // could not find, and running it twice would make two. + let resolved = MergeOperator::resolved_props( + Some(literal_properties), + self.property_exprs.get(i).and_then(|e| e.as_ref()), + &record, + store, + )? + .unwrap_or_default(); + let properties = &resolved; let source_id = match record.get(source_var).and_then(|v| v.node_id()) { Some(id) => id, None => continue, @@ -17183,12 +17212,16 @@ impl PhysicalOperator for MatchMergeEdgeOperator { let mut parts: Vec = self .edges_to_merge .iter() - .map(|(source, target, edge_type, properties, edge_var, undirected)| { + .enumerate() + .map(|(i, (source, target, edge_type, properties, edge_var, undirected))| { let edge = format!( "[{}:{}{}]", edge_var.as_deref().unwrap_or(""), edge_type.as_str(), - format_properties(Some(properties), None) + format_properties( + Some(properties), + self.property_exprs.get(i).and_then(|e| e.as_ref()) + ) ); if *undirected { format!("({})-{}-({})", source, edge, target) diff --git a/src/query/executor/planner.rs b/src/query/executor/planner.rs index 001dfef7..147c2101 100644 --- a/src/query/executor/planner.rs +++ b/src/query/executor/planner.rs @@ -9399,6 +9399,8 @@ impl QueryPlanner { // Extract edge patterns from MERGE clause let mut edges_to_merge = Vec::new(); + // Each edge's non-literal properties, by position in `edges_to_merge`. + let mut edge_property_exprs = Vec::new(); // `MERGE p = (a)-[:R]->(b)` binds `p` (#876). An anonymous // relationship inside a named path is given a synthetic handle, for // the same reason `CREATE` gives one to an anonymous node: the path @@ -9447,6 +9449,7 @@ impl QueryPlanner { edge_var, undirected, )); + edge_property_exprs.push(edge.property_exprs.clone()); } current_var = target_var; } @@ -9495,7 +9498,8 @@ impl QueryPlanner { .collect(); operator = Box::new( MatchMergeEdgeOperator::new(operator, edges_to_merge, on_create, on_match) - .with_entity_sets(on_create_entity, on_match_entity), + .with_entity_sets(on_create_entity, on_match_entity) + .with_property_exprs(edge_property_exprs), ); if !merge_named_paths.is_empty() { operator = Box::new(crate::query::executor::operator::BindPathOperator::new( diff --git a/tests/merge_row_properties.rs b/tests/merge_row_properties.rs index 4115fa8b..2e007055 100644 --- a/tests/merge_row_properties.rs +++ b/tests/merge_row_properties.rs @@ -101,3 +101,98 @@ fn merge_between_two_bound_nodes_still_reuses_them() { assert_eq!(count(&store, "MATCH (n:B) RETURN n"), 1); assert_eq!(count(&store, "MATCH ()-[r:R]->() RETURN r"), 1); } + +// --------------------------------------------------------------------------- +// The relationship half: a MERGE that wires an edge between endpoints a MATCH +// already bound, keyed on a value from the row. +// +// ```cypher +// UNWIND $rows AS r MATCH (s {id: r.src}), (d {id: r.dst}) MERGE (s)-[:T {id: r.id}]->(d) +// ``` +// +// is the bulk edge-upsert idiom, and it goes to `MatchMergeEdgeOperator`, +// which was only ever handed the literal half of the property map. The +// expressions were dropped: MERGE matched any `:T` between the pair, and when +// it created one, created it without the property. Two rows for one pair wrote +// ONE edge with no `id`, and the statement reported success. Found loading the +// INT-10 reference pipeline, where 60 transfers became 55 edges (#1715). +// --------------------------------------------------------------------------- + +#[test] +fn an_edge_keyed_on_the_row_is_one_edge_per_distinct_key_between_bound_nodes() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (:A {id: 1}), (:B {id: 1})"); + let upsert = "UNWIND [{id: 't1'}, {id: 't2'}, {id: 't1'}] AS r \ + MATCH (a:A {id: 1}), (b:B {id: 1}) MERGE (a)-[:T {id: r.id}]->(b)"; + run(&mut store, upsert); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "'t1' twice is still one edge"); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't1'}]->() RETURN t"), 1, "the key was written"); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't2'}]->() RETURN t"), 1); + + run(&mut store, upsert); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "still idempotent"); +} + +#[test] +fn an_edge_with_a_different_key_is_not_a_match() { + // The silent half of the defect: the existing edge was "matched" whatever + // its key, so a later SET overwrote it instead of creating a second. + let mut store = GraphStore::new(); + run(&mut store, "CREATE (a:A {id: 1})-[:T {id: 'old', amount: 1}]->(b:B {id: 1})"); + run( + &mut store, + "UNWIND [{id: 'new', amount: 2}] AS r MATCH (a:A {id: 1}), (b:B {id: 1}) \ + MERGE (a)-[t:T {id: r.id}]->(b) SET t.amount = r.amount", + ); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 'old', amount: 1}]->() RETURN t"), 1, "untouched"); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 'new', amount: 2}]->() RETURN t"), 1); +} + +#[test] +fn an_edge_can_key_on_a_property_of_another_bound_node() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (:A {id: 1}), (:B {id: 1}), (:Src {k: 'x'}), (:Src {k: 'y'})"); + let merge = "MATCH (s:Src) MATCH (a:A {id: 1}), (b:B {id: 1}) MERGE (a)-[:T {k: s.k}]->(b)"; + run(&mut store, merge); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2); + assert_eq!(count(&store, "MATCH ()-[t:T {k: 'x'}]->() RETURN t"), 1); + assert_eq!(count(&store, "MATCH ()-[t:T {k: 'y'}]->() RETURN t"), 1); +} + +#[test] +fn literal_and_row_properties_on_one_edge_must_both_match() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (a:A {id: 1})-[:T {kind: 'other', id: 't1'}]->(b:B {id: 1})"); + let merge = "UNWIND ['t1'] AS x MATCH (a:A {id: 1}), (b:B {id: 1}) \ + MERGE (a)-[:T {kind: 'wire', id: x}]->(b)"; + run(&mut store, merge); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "kind differs, so no match"); + assert_eq!(count(&store, "MATCH ()-[t:T {kind: 'wire', id: 't1'}]->() RETURN t"), 1); +} + +#[test] +fn on_create_and_on_match_see_the_row_keyed_edge() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (:A {id: 1}), (:B {id: 1})"); + let merge = "UNWIND ['t1'] AS x MATCH (a:A {id: 1}), (b:B {id: 1}) \ + MERGE (a)-[t:T {id: x}]->(b) ON CREATE SET t.created = true ON MATCH SET t.matched = true"; + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't1', created: true}]->() RETURN t"), 1); + assert_eq!(count(&store, "MATCH ()-[t:T]->() WHERE t.matched IS NULL RETURN t"), 1); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't1', matched: true}]->() RETURN t"), 1); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 1); +} + +#[test] +fn an_undirected_edge_keyed_on_the_row_matches_either_way_round() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (a:A {id: 1})<-[:T {id: 't1'}]-(b:B {id: 1})"); + let merge = "UNWIND ['t1', 't2'] AS x MATCH (a:A {id: 1}), (b:B {id: 1}) MERGE (a)-[:T {id: x}]-(b)"; + run(&mut store, merge); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "t1 existed reversed; only t2 is new"); +} From 3fed006823a1304df3a241419a7350809e97de21 Mon Sep 17 00:00:00 2001 From: Raghuvarma905 Date: Tue, 20 Oct 2026 00:34:49 +0530 Subject: [PATCH 2/4] Reference Airflow and Dagster pipelines, run against a server in CI (INT-10, #1715) INT-10 asks for a documented, tested integration with Airflow/Dagster-style orchestration: one reference pipeline per tool. Nothing in the repository mentioned either. examples/orchestration/ holds both, sharing one module for what a step does, so the two cannot drift: samyama_pipeline.py the load: constraint, accounts, transfers, and validation against the source files; standard library only samyama_airflow_dag.py Airflow 3 DAG; server and token from an Airflow Connection, files and batch size as DAG params samyama_dagster.py Dagster assets with blocking asset checks; server from SAMYAMA_URL / SAMYAMA_TOKEN via EnvVar sample/ synthetic accounts and transfers (not data/, which .gitignore swallows) What the load guarantees, and how: - A retry is safe: every write is a MERGE keyed on the source id, behind a unique constraint. Both suites run each pipeline twice. - Values are bound in `params`, never interpolated. The Python SDK's query() takes no parameters, which is why this uses HTTP. - Only an unavailable server (no connection, timeout, 5xx, 408, 429) is retried. Refused statements, a refused token, wrong input and a failed validation fail at once (AirflowFailException / Failure(allow_retries= False)). - A transfer naming an account the graph does not hold is refused before anything is written; otherwise the MATCH finds nothing and the row is skipped in silence. Two engine behaviours shaped the Cypher and are written down in docs/ORCHESTRATION.md: two MATCH clauses after UNWIND plan as a cartesian product (4.5 ms vs a 281 s timeout at 1,000 accounts, so the pipeline uses one comma-separated MATCH), and a non-literal property map on an anonymous node is refused in MATCH. Tests (examples/orchestration/tests/, 52): each test starts its own release server on spare ports with a fresh data directory. The Airflow DAG runs through dag.test(), the Dagster assets through materialize. A missing binary is an error, not a skip. A new `orchestration` job in ci.yml runs them against the binary `test` already builds; it is listed in UNGATED in tests/ci_gate_enumeration.rs rather than in ci-ok's needs, because making a PyPI-installing suite block every merge is a separate decision. Docs: ORCHESTRATION.md (design, guarantees, limits), and a how-to per tool. Every command in the how-tos was run as written against a local server with an auth file; doing so found that they omitted copying sample/, now fixed. test_docs.py checks each Cypher block shown is one the pipeline runs. --- .github/workflows/ci.yml | 67 ++++ docs/ORCHESTRATION-AIRFLOW.md | 128 +++++++ docs/ORCHESTRATION-DAGSTER.md | 113 ++++++ docs/ORCHESTRATION.md | 151 ++++++++ docs/README.md | 5 +- examples/README.md | 13 +- examples/orchestration/README.md | 16 + examples/orchestration/requirements.txt | 12 + examples/orchestration/sample/accounts.csv | 25 ++ examples/orchestration/sample/transfers.csv | 61 +++ examples/orchestration/samyama_airflow_dag.py | 131 +++++++ examples/orchestration/samyama_dagster.py | 140 +++++++ examples/orchestration/samyama_pipeline.py | 359 ++++++++++++++++++ examples/orchestration/tests/conftest.py | 199 ++++++++++ examples/orchestration/tests/test_airflow.py | 111 ++++++ examples/orchestration/tests/test_dagster.py | 117 ++++++ examples/orchestration/tests/test_docs.py | 43 +++ examples/orchestration/tests/test_pipeline.py | 268 +++++++++++++ tests/ci_gate_enumeration.rs | 17 +- 19 files changed, 1968 insertions(+), 8 deletions(-) create mode 100644 docs/ORCHESTRATION-AIRFLOW.md create mode 100644 docs/ORCHESTRATION-DAGSTER.md create mode 100644 docs/ORCHESTRATION.md create mode 100644 examples/orchestration/README.md create mode 100644 examples/orchestration/requirements.txt create mode 100644 examples/orchestration/sample/accounts.csv create mode 100644 examples/orchestration/sample/transfers.csv create mode 100644 examples/orchestration/samyama_airflow_dag.py create mode 100644 examples/orchestration/samyama_dagster.py create mode 100644 examples/orchestration/samyama_pipeline.py create mode 100644 examples/orchestration/tests/conftest.py create mode 100644 examples/orchestration/tests/test_airflow.py create mode 100644 examples/orchestration/tests/test_dagster.py create mode 100644 examples/orchestration/tests/test_docs.py create mode 100644 examples/orchestration/tests/test_pipeline.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 44052c97..05de395d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -897,6 +897,73 @@ jobs: ${{ runner.temp }}/gpml-conformance/suite/results/summary.json if-no-files-found: warn + # INT-10 asks for an orchestration integration that is documented AND tested: + # one reference pipeline per tool, run against a server. This runs both -- + # the Airflow DAG through `dag.test()` and the Dagster assets through + # `materialize` -- against the release binary `test` built, each test on its + # own server with a fresh data directory (examples/orchestration/tests/). + # + # A reference pipeline that is not run rots exactly like documentation does, + # only later: it keeps importing cleanly long after the HTTP API it calls has + # moved, and the first person to find out is the one who copied it. + # + # Not in `ci-ok`'s `needs`, for the reason `gpml` is not: making a new suite + # block every merge -- here, one that installs Airflow from PyPI -- is a policy + # decision, separate from adding the suite. tests/ci_gate_enumeration.rs + # records the exemption so it cannot be forgotten. + orchestration: + name: orchestration (Airflow and Dagster reference pipelines) + needs: [changes, test] + if: needs.changes.outputs.code == 'true' + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: {python-version: "3.12"} + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + + # The binary `test` already built. A missing artifact is not a failure: + # the next step builds the engine instead, as the gpml job does. + - name: Take the release binary from the test job + continue-on-error: true + uses: actions/download-artifact@v4 + with: + name: samyama-release + path: target/release + + - name: Use it if it arrived, otherwise build it + run: | + set -euo pipefail + if [ -f target/release/samyama ]; then + # An artifact download does not preserve the executable bit. + chmod +x target/release/samyama + echo "using the binary built by the test job" + else + echo "::notice::no release artifact; building the engine" + cargo build --release --bin samyama + fi + + - name: Install Airflow and Dagster at the tested versions + working-directory: examples/orchestration + # Airflow's constraints file is per Airflow version and per Python + # version; both are read rather than restated, so bumping the pin in + # requirements.txt cannot leave this step installing against the old + # constraints. + run: | + set -euo pipefail + airflow=$(sed -n 's/^apache-airflow==//p' requirements.txt) + py=$(python -c 'import sys; print(f"{sys.version_info[0]}.{sys.version_info[1]}")') + pip install -r requirements.txt \ + -c "https://raw.githubusercontent.com/apache/airflow/constraints-${airflow}/constraints-${py}.txt" + + - name: Run both pipelines against the engine + working-directory: examples/orchestration + env: + SAMYAMA_BIN: ${{ github.workspace }}/target/release/samyama + run: python -m pytest -v tests/ + # The one check branch protection should require. # # `test` and the two snapshot jobs take their check names from a matrix. When an diff --git a/docs/ORCHESTRATION-AIRFLOW.md b/docs/ORCHESTRATION-AIRFLOW.md new file mode 100644 index 00000000..e682875b --- /dev/null +++ b/docs/ORCHESTRATION-AIRFLOW.md @@ -0,0 +1,128 @@ +# Loading Samyama from Apache Airflow + +How to run the reference DAG, +[`examples/orchestration/samyama_airflow_dag.py`](../examples/orchestration/samyama_airflow_dag.py), +against your server and your files. What the pipeline does, the guarantees it +makes and what it does not do are in [ORCHESTRATION.md](./ORCHESTRATION.md). + +Tested with Airflow 3.3.2 on Python 3.12. Airflow 2 is not supported: the DAG +uses Airflow 3's `airflow.sdk`. + +--- + +## 1. A credential for the pipeline + +Skip this if your server runs without `--auth-file`. + +```bash +samyama auth-token pipeline +``` + +prints a line for the credential file and, once, the token. Add the line to +the file the server reads with `:roles=write` on the end -- a pipeline that +loads data has no business managing tenants or restoring snapshots: + +```text +pipeline::roles=write +``` + +Keep the token for step 3. The server stores only its digest. + +## 2. Install, and put the files in the DAGs folder + +```bash +pip install "apache-airflow==3.3.2" \ + -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt + +cp examples/orchestration/samyama_airflow_dag.py \ + examples/orchestration/samyama_pipeline.py \ + "$AIRFLOW_HOME/dags/" +cp -r examples/orchestration/sample "$AIRFLOW_HOME/dags/" +``` + +Both `.py` files: the DAG imports `samyama_pipeline`, and Airflow puts the DAGs +folder on `sys.path`. `sample/` is the default input, found next to +`samyama_pipeline.py`; leave it out if every run will name its own files, and a +run that does not fails at once with `no such file`. Nothing else is needed -- +`samyama_pipeline` is standard library only, and there is no Samyama provider +package to install. + +## 3. A Connection to the server + +The DAG reads the server from an Airflow Connection, `samyama_default` unless a +run says otherwise. `schema` is the protocol, the convention Airflow's own HTTP +hook uses; the token, if any, goes in `password`: + +```bash +airflow connections add samyama_default \ + --conn-type http \ + --conn-host graph.internal \ + --conn-port 8080 \ + --conn-schema https \ + --conn-password "$SAMYAMA_TOKEN" +``` + +Any secrets backend Airflow supports works the same way, because the DAG asks +Airflow for the Connection rather than reading a variable of its own. As an +environment variable: + +```bash +export AIRFLOW_CONN_SAMYAMA_DEFAULT='{"conn_type": "http", "host": "graph.internal", "port": 8080, "schema": "https", "password": "..."}' +``` + +Optional: `{"timeout_seconds": 60}` in the Connection's extras sets how long one +statement may take. The default is 30. + +## 4. Run it + +Once, without the scheduler, to see it work -- with the bundled sample: + +```bash +airflow dags test samyama_accounts_transfers +``` + +With your own files: + +```bash +airflow dags test samyama_accounts_transfers \ + -c '{"accounts_csv": "/data/accounts.csv", "transfers_csv": "/data/transfers.csv"}' +``` + +The same JSON is what you pass to `airflow dags trigger --conf`, or type into +the *Trigger DAG* form in the UI. + +| Param | Default | | +|---|---|---| +| `samyama_conn_id` | `samyama_default` | the Connection to use | +| `accounts_csv` | the bundled sample | columns `id, holder, country, opened` | +| `transfers_csv` | the bundled sample | columns `id, src, dst, amount, at` | +| `batch_size` | `500` | rows per statement | + +The DAG has `schedule=None`: it runs when triggered. Give it a schedule, or an +asset to wait on, when you deploy it -- that choice belongs to the deployment, +not to a reference. + +## 5. When a run fails + +The failing task's log names the cause. Two kinds: + +- **Retried, with backoff, up to three times**: the server could not be + reached, timed out, or answered 5xx. +- **Failed at once, no retry**: the server refused the statement or the token, + the input is wrong (the message names the file and line), or validation found + the graph disagrees with the files (the message names each check, what it + expected and what it found). + +A failed run can leave part of a load in the graph. Fix the cause and run it +again: every write is an upsert keyed on the source id, so the second run +completes the first rather than duplicating it. + +## How this page is tested + +`examples/orchestration/tests/test_airflow.py` runs the DAG through +`dag.test()` against a real server: a full load, a second run that must leave +the graph unchanged, a file naming an unknown account (one try, nothing +written), a token taken from the Connection, a missing token (one try), and a +Connection named by the `samyama_conn_id` param. The commands above were run as +written against a local server before this page was published; they are not +re-run by CI. diff --git a/docs/ORCHESTRATION-DAGSTER.md b/docs/ORCHESTRATION-DAGSTER.md new file mode 100644 index 00000000..06111650 --- /dev/null +++ b/docs/ORCHESTRATION-DAGSTER.md @@ -0,0 +1,113 @@ +# Loading Samyama from Dagster + +How to run the reference assets, +[`examples/orchestration/samyama_dagster.py`](../examples/orchestration/samyama_dagster.py), +against your server and your files. What the pipeline does, the guarantees it +makes and what it does not do are in [ORCHESTRATION.md](./ORCHESTRATION.md). + +Tested with Dagster 1.13.26 on Python 3.12. + +--- + +## 1. A credential for the pipeline + +Skip this if your server runs without `--auth-file`. + +```bash +samyama auth-token pipeline +``` + +prints a line for the credential file and, once, the token. Add the line to +the file the server reads with `:roles=write` on the end -- a pipeline that +loads data has no business managing tenants or restoring snapshots: + +```text +pipeline::roles=write +``` + +## 2. Install, and put the files in your code location + +```bash +pip install "dagster==1.13.26" "dagster-webserver==1.13.26" +cp examples/orchestration/samyama_dagster.py examples/orchestration/samyama_pipeline.py my_project/ +cp -r examples/orchestration/sample my_project/ +``` + +Both `.py` files: `samyama_dagster` imports `samyama_pipeline`, which is +standard library only. `sample/` is the default input, found next to +`samyama_pipeline.py`; leave it out if every run will name its own files. There +is no Samyama integration package to install. + +## 3. Point it at the server + +| Variable | | +|---|---| +| `SAMYAMA_URL` | required -- the server's HTTP endpoint, e.g. `https://graph.internal:8080` | +| `SAMYAMA_TOKEN` | the token from step 1; leave unset for a server without an auth file | + +Both are read through `dg.EnvVar`, so the token is resolved when a run starts +and is not stored with the code location or shown in the UI. + +## 4. Run it + +```bash +export SAMYAMA_URL=http://localhost:8080 +cd my_project +dagster dev -f samyama_dagster.py +``` + +and materialize the `samyama` group from the UI, or without the UI: + +```bash +dagster asset materialize -f samyama_dagster.py --select '*' +``` + +The source files and batch size are the `source_files` resource. They default +to the bundled sample; point a run at your own in the launchpad's config: + +```yaml +resources: + source_files: + config: + accounts_csv: /data/accounts.csv + transfers_csv: /data/transfers.csv + batch_size: 500 +``` + +| Asset | Depends on | Check | +|---|---|---| +| `samyama_schema` | | | +| `accounts` | `samyama_schema` | `accounts_match_source` | +| `transfers` | `accounts` | `transfers_match_source` | + +The checks are **blocking**: if `accounts_match_source` fails, `transfers` is +not loaded on top of accounts that do not match their file. Each check's +metadata records, per condition, how many it expected and how many it found. + +No schedule or sensor is defined; add one when you deploy it -- that choice +belongs to the deployment, not to a reference. + +## 5. When a run fails + +- **Retried, with exponential backoff, up to three times**: the server could + not be reached, timed out, or answered 5xx. +- **Failed at once, no retry** (`Failure(allow_retries=False)`): the server + refused the statement or the token, or the input is wrong -- the message + names the file and line. +- **A check failed**: the graph disagrees with the files. The check's + description names each condition that failed. + +A failed run can leave part of a load in the graph. Fix the cause and +materialize again: every write is an upsert keyed on the source id, so the +second run completes the first rather than duplicating it. + +## How this page is tested + +`examples/orchestration/tests/test_dagster.py` materializes the assets and +checks against a real server: a full load with both checks passing, a second +run that must leave the graph unchanged, a file naming an unknown account (the +step fails, is not retried, writes nothing), a failing accounts check that must +stop `transfers` from loading, a token from the resource, a missing token, and +the shipped definitions reading `SAMYAMA_URL` and `SAMYAMA_TOKEN` from the +environment. The commands above were run as written against a local server +before this page was published; they are not re-run by CI. diff --git a/docs/ORCHESTRATION.md b/docs/ORCHESTRATION.md new file mode 100644 index 00000000..0263eeed --- /dev/null +++ b/docs/ORCHESTRATION.md @@ -0,0 +1,151 @@ +# Loading Samyama from Airflow and Dagster + +A scheduled pipeline that loads files into the graph and checks the result, in +the two orchestrators most data teams already run. One reference pipeline per +tool, both run against a real server by the test suite: + +| Tool | Reference | How-to | +|---|---|---| +| Apache Airflow 3 | [`examples/orchestration/samyama_airflow_dag.py`](../examples/orchestration/samyama_airflow_dag.py) | [ORCHESTRATION-AIRFLOW.md](./ORCHESTRATION-AIRFLOW.md) | +| Dagster | [`examples/orchestration/samyama_dagster.py`](../examples/orchestration/samyama_dagster.py) | [ORCHESTRATION-DAGSTER.md](./ORCHESTRATION-DAGSTER.md) | + +Both are thin. What a step *does* is in +[`samyama_pipeline.py`](../examples/orchestration/samyama_pipeline.py), once, so +the two cannot drift into loading the same files two different ways. It is +standard library only, so it adds nothing to an orchestrator's worker image. + +--- + +## What the pipeline does + +``` +schema -> accounts -> transfers -> validate +``` + +| Step | What it does | +|---|---| +| schema | A unique constraint on `:Account(id)`. It backs the id lookups with an index, and makes the engine refuse a second account with a taken id. `IF NOT EXISTS`, so repeating it is harmless. | +| accounts | Reads `accounts.csv` (`id, holder, country, opened`) and upserts one `:Account` per row, in batches. | +| transfers | Reads `transfers.csv` (`id, src, dst, amount, at`), checks every account it names exists, then upserts one `:TRANSFER` relationship per row, in batches. | +| validate | Reads the files again and asks the graph whether it holds exactly them. Fails the run if not. | + +The sample in `examples/orchestration/sample/` is synthetic: 24 accounts and 60 +transfers, generated, describing no real party. + +## The three properties every orchestrated load needs + +**A retry is safe.** Orchestrators retry on their own initiative -- a worker +restart, a timeout, a node drained for maintenance. Every write is a `MERGE` +keyed on the source row's id, so running a step twice leaves the graph as +running it once. Both test suites run each pipeline twice and compare the graph +with the files. + +**Values are bound, never pasted into the statement.** Rows travel as the +`params` of `POST /api/query` (#1463): + +```cypher +UNWIND $rows AS r +MATCH (s:Account {id: r.src}), (d:Account {id: r.dst}) +MERGE (s)-[t:TRANSFER {id: r.id}]->(d) +SET t.amount = r.amount, t.at = r.at +RETURN count(t) AS written +``` + +A holder name with a quote in it, or text that looks like Cypher, is stored as +the string it is; the tests load a few such values and read them back. This is +also why the pipelines talk HTTP rather than use the Python SDK: the SDK's +`query()` takes no parameters, so using it would mean building statement text +out of file contents. + +**A failure says whether retrying can help.** + +| Failure | Retried? | Why | +|---|---|---| +| Server unreachable, timeout, any 5xx (including 503 when the server refuses writes after a failed flush) | Yes, with backoff | It can clear on its own. | +| The server refused the statement or the credential (4xx) | No | It will refuse it again. | +| The input is wrong: a missing column, an empty field, a duplicate id, an amount that is not a number, a transfer naming an account that does not exist | No | A retry reads the same file. | +| Validation found the graph disagrees with the files | No | A retry re-checks the same graph. | + +Retrying a permanent failure only delays the page to whoever has to fix it. +Airflow gets `AirflowFailException`; Dagster gets `Failure(allow_retries=False)`. + +## What validation checks + +Nothing it checks is taken from the load step's own report -- the files are read +again and the graph is asked: + +| Check | Catches | +|---|---| +| every account, found by id, holds the file's `holder`, `country`, `opened` | a missed row, a value that did not arrive, a value something else changed | +| no account id is held by two nodes | a writer that used `CREATE` where it should `MERGE` (the constraint stops this once the schema step has run; the check covers graphs where it has not) | +| every transfer runs between the accounts the file names and holds its `id`, `amount`, `at` | a dropped or misdirected edge, a changed value | +| no transfer id from this file is held by two relationships | a transfer whose account changed between two runs: `MERGE` matches on both endpoints and the id, so it adds a second relationship rather than moving the first | + +The transfer step also refuses, before writing anything, a file that names an +account the graph does not hold. Without that, the `MATCH` would find no row for +it, the `MERGE` would never run, and the statement would succeed having silently +skipped the transfer. + +## Adapting it to your data + +Copy `samyama_pipeline.py` and change the statements and the two readers; the +orchestrator files need to change only if you add or rename a step. Keep: + +- **`MERGE` on a key that comes from the source**, with a unique constraint on + it. Without a key a retry duplicates. +- **Both endpoints of a relationship in one comma-separated `MATCH`.** Written + as two `MATCH` clauses -- which means the same thing -- the planner does not + look either up by id: it pairs every node with every node and filters. On a + graph of 1,000 accounts a 200-row batch took 4.5 ms in the comma form and + timed out after 281 s in the two-clause form (2026-10-19, release build of + `dev` at `23a96ec`, one laptop, one run each -- the gap is the point, not the + figures). +- **Named variables on nodes whose properties come from the row.** + `(:Account {id: r.src})` with no variable is refused in a `MATCH`; write + `(s:Account {id: r.src})`. +- **Batches.** The default is 500 rows per statement; it is a parameter of both + pipelines. + +## What this does not do + +- **A run is not one transaction.** Each batch commits on its own, so while a + run is in progress -- or after it fails -- readers see a partial load, which + the next run completes. `POST /api/tx/begin` can make a run atomic, but a + transaction holds the store's write lock: every other reader and writer waits + for the whole load, and the transaction is rolled back if it outlives the + session timeout. That trade is not made for you here. +- **Deletions are not propagated.** A row removed from the file stays in the + graph. Upserts cannot tell "deleted" from "not in this extract". +- **One graph per server.** The open-source server serves the graph named + `default`; use a separate server and data directory per dataset. +- **dbt is not covered.** dbt compiles SQL for a SQL warehouse, and there is no + dbt adapter for Cypher. +- **Airflow 2 is not supported.** The DAG imports from `airflow.sdk`, which is + Airflow 3's authoring interface. + +## How it is tested + +`examples/orchestration/tests/`, run by the `orchestration` job in +`.github/workflows/ci.yml` against the release binary the `test` job builds: + +```bash +cargo build --release --bin samyama +cd examples/orchestration +pip install -r requirements.txt \ + -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt +python -m pytest -v tests/ +``` + +Each test starts its own server on spare ports with a fresh data directory, so +no test sees another's graph and nothing touches a server you are running. +`SAMYAMA_BIN` names another binary. + +What the tests run: the Airflow DAG through `dag.test()` and the Dagster assets +through `materialize`, in process, with Airflow's own Connection lookup and +Dagster's own resources. What they do **not** run: a deployed scheduler with +separate workers, or any version other than the pinned Airflow 3.3.2 and +Dagster 1.13.26. + +The CI job is not yet one of the checks `ci-ok` waits for, so a failure there +shows on a pull request without blocking its merge; whether it should block is +a separate decision, recorded in `tests/ci_gate_enumeration.rs`. diff --git a/docs/README.md b/docs/README.md index 4e89cc00..ca27d174 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,6 +1,6 @@ # Samyama Graph — Documentation Index -**Last updated:** 2026-10-14 · **Engine version:** see [Releases](https://github.com/samyama-ai/samyama-graph/releases) — documents below state the version they were verified against +**Last updated:** 2026-10-19 · **Engine version:** see [Releases](https://github.com/samyama-ai/samyama-graph/releases) — documents below state the version they were verified against ## Primary sources @@ -47,6 +47,9 @@ eight of eighteen documents (#1450). - **[ACID-GUARANTEES.md](./ACID-GUARANTEES.md)** — the transaction model: MVCC, the logical WAL, what durability `SAMYAMA_FSYNC` does and does not buy, and why there is no Raft replication to rely on. - **[FAILURE-MODES.md](./FAILURE-MODES.md)** — how the engine behaves when things go wrong. - **[BI-CONNECTIVITY.md](./BI-CONNECTIVITY.md)** — connecting BI tools. +- **[ORCHESTRATION.md](./ORCHESTRATION.md)** — loading the graph from a scheduled pipeline: a retry-safe, parameter-bound load with validation, as reference pipelines for Airflow and Dagster that the test suite runs against a real server. +- **[ORCHESTRATION-AIRFLOW.md](./ORCHESTRATION-AIRFLOW.md)** — how to run the reference DAG in Airflow 3: credential, Connection, params, what a failure means. +- **[ORCHESTRATION-DAGSTER.md](./ORCHESTRATION-DAGSTER.md)** — how to run the reference assets and checks in Dagster. ### Data, provenance and privacy diff --git a/examples/README.md b/examples/README.md index f801b6d4..0ed2f351 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,7 +1,7 @@ # `examples/` Every program in this directory, what it does, and what data it needs. -`cargo run --release --example ` runs any of the `.rs` files (the two non-Rust +`cargo run --release --example ` runs any of the `.rs` files (the four non-Rust programs say how below). The names are an interface: `scripts/run_all_examples.sh`, `scripts/verify-sweep.sh`, `scripts/regression-test.sh`, `.github/workflows/ci.yml` and external knowledge-graph repositories invoke them by name, so they do not get renamed. @@ -9,7 +9,7 @@ external knowledge-graph repositories invoke them by name, so they do not get re **Start here** if you are new: the demos on built-in data. They need nothing but a checkout. -**Most of this directory is internal.** Of the 127 programs, 60 are performance +**Most of this directory is internal.** Of the 129 programs, 60 are performance investigations, conformance checks or maintainer tools — written to answer one question about the engine, usually once, often to kill a hypothesis. They print verdicts and timings, not results you can use. Those sections say *Internal* in their opening line; @@ -29,6 +29,7 @@ Purposes below are taken from each file's own header comment. | [Conformance and correctness checks](#conformance-and-correctness-checks) | 32 | No - internal | | [Maintainer tools](#maintainer-tools) | 3 | No - internal | | [Hardware client sketch](#hardware-client-sketch) | 1 | No - internal | +| [Orchestration reference pipelines](#orchestration-reference-pipelines) | 2 | Yes | Helper code shared between programs lives in `examples/common/`, which holds no runnable program. A dataset's loader and its benchmarks share `common/_common.rs`, @@ -235,6 +236,14 @@ not need them. |---|---|---| | `esp32/samyama_esp32.ino` | Writes to a Samyama Edge node from an ESP32 over RESP (API-15). RESP rather than HTTP because a TLS stack, a JSON encoder and a chunked reader are most of an ESP32's flash and all of its RAM headroom. Nothing here heap-allocates. | Arduino/ESP32 toolchain and a reachable Samyama RESP endpoint. | +## Orchestration reference pipelines +Not Cargo examples. Python, run by Airflow or Dagster against a running server; both share the load steps in `orchestration/samyama_pipeline.py`. How to run them: [`docs/ORCHESTRATION.md`](../docs/ORCHESTRATION.md). `orchestration/tests/` runs both against a real server, and the `orchestration` CI job runs those tests. + +| Program | What it does | Data it needs | +|---|---|---| +| `orchestration/samyama_airflow_dag.py` | Airflow 3 DAG (INT-10): constraint, then accounts and transfers upserted in batches with bound parameters, then a validation task that compares the graph with the files. Retries only failures a retry can fix. | `orchestration/sample/` (synthetic, bundled), or your own CSVs as DAG params; a server named by an Airflow Connection. | +| `orchestration/samyama_dagster.py` | The same load as Dagster assets, with blocking asset checks: transfers are not loaded on top of accounts that fail theirs. | `orchestration/sample/`, or your own CSVs in the `source_files` resource; `SAMYAMA_URL` and optionally `SAMYAMA_TOKEN`. | + ## Terms used in the file headers Several terms appear in these headers with no definition in this repository. Each one is diff --git a/examples/orchestration/README.md b/examples/orchestration/README.md new file mode 100644 index 00000000..0aa34c74 --- /dev/null +++ b/examples/orchestration/README.md @@ -0,0 +1,16 @@ +# Orchestration reference pipelines + +Loading Samyama from Airflow and Dagster (INT-10). Start with +[`docs/ORCHESTRATION.md`](../../docs/ORCHESTRATION.md). + +| File | | +|---|---| +| `samyama_pipeline.py` | The load steps both pipelines run. Standard library only. | +| `samyama_airflow_dag.py` | Airflow 3 DAG — [how to run it](../../docs/ORCHESTRATION-AIRFLOW.md). | +| `samyama_dagster.py` | Dagster assets and checks — [how to run them](../../docs/ORCHESTRATION-DAGSTER.md). | +| `sample/` | Synthetic accounts and transfers, the default input. | +| `requirements.txt` | The orchestrator versions the tests were run against. | +| `tests/` | Runs both pipelines against a real server: `python -m pytest tests/`. | + +The sample is not in `data/` because the repository ignores every directory of +that name. diff --git a/examples/orchestration/requirements.txt b/examples/orchestration/requirements.txt new file mode 100644 index 00000000..acc6684d --- /dev/null +++ b/examples/orchestration/requirements.txt @@ -0,0 +1,12 @@ +# The versions tests/ were last run against. The DAG and the assets use only +# stable public API, but a version not listed here has not been tested. +# +# Airflow wants its constraints file, which pins its transitive dependencies: +# pip install -r requirements.txt \ +# -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt +# +# Each pipeline needs only its own orchestrator; `samyama_pipeline.py` is +# standard library only. +apache-airflow==3.3.2 +dagster==1.13.26 +pytest diff --git a/examples/orchestration/sample/accounts.csv b/examples/orchestration/sample/accounts.csv new file mode 100644 index 00000000..2612a23c --- /dev/null +++ b/examples/orchestration/sample/accounts.csv @@ -0,0 +1,25 @@ +id,holder,country,opened +ACC-0001,Holder 01,DE,2021-11-11 +ACC-0002,Holder 02,GB,2023-07-24 +ACC-0003,Holder 03,IN,2022-06-30 +ACC-0004,Holder 04,GB,2019-05-19 +ACC-0005,Holder 05,GB,2021-06-02 +ACC-0006,Holder 06,IN,2022-08-24 +ACC-0007,Holder 07,SG,2023-06-13 +ACC-0008,Holder 08,IN,2021-09-28 +ACC-0009,Holder 09,AE,2022-12-15 +ACC-0010,Holder 10,DE,2021-10-08 +ACC-0011,Holder 11,US,2020-04-21 +ACC-0012,Holder 12,US,2023-10-04 +ACC-0013,Holder 13,IN,2019-03-12 +ACC-0014,Holder 14,DE,2023-07-20 +ACC-0015,Holder 15,DE,2021-07-10 +ACC-0016,Holder 16,SG,2022-02-11 +ACC-0017,Holder 17,GB,2019-10-21 +ACC-0018,Holder 18,IN,2021-11-13 +ACC-0019,Holder 19,IN,2022-03-04 +ACC-0020,Holder 20,DE,2021-06-19 +ACC-0021,Holder 21,AE,2020-04-19 +ACC-0022,Holder 22,US,2023-07-20 +ACC-0023,Holder 23,IN,2022-04-07 +ACC-0024,Holder 24,AE,2020-09-10 diff --git a/examples/orchestration/sample/transfers.csv b/examples/orchestration/sample/transfers.csv new file mode 100644 index 00000000..9370975b --- /dev/null +++ b/examples/orchestration/sample/transfers.csv @@ -0,0 +1,61 @@ +id,src,dst,amount,at +TX-00024,ACC-0014,ACC-0004,10626.50,2026-09-01T17:26:00 +TX-00008,ACC-0006,ACC-0022,2040.19,2026-09-01T22:51:00 +TX-00023,ACC-0007,ACC-0004,22974.72,2026-09-02T04:31:00 +TX-00046,ACC-0011,ACC-0021,4836.15,2026-09-03T13:20:00 +TX-00003,ACC-0019,ACC-0021,15383.89,2026-09-05T10:03:00 +TX-00034,ACC-0008,ACC-0001,13521.34,2026-09-05T10:10:00 +TX-00007,ACC-0003,ACC-0009,16470.95,2026-09-05T11:15:00 +TX-00001,ACC-0002,ACC-0008,15511.29,2026-09-07T01:34:00 +TX-00038,ACC-0008,ACC-0014,24447.60,2026-09-07T17:05:00 +TX-00016,ACC-0002,ACC-0001,1757.14,2026-09-08T07:38:00 +TX-00013,ACC-0005,ACC-0010,18480.51,2026-09-08T13:21:00 +TX-00014,ACC-0003,ACC-0024,8392.61,2026-09-11T14:42:00 +TX-00055,ACC-0004,ACC-0020,19788.88,2026-09-12T05:15:00 +TX-00053,ACC-0007,ACC-0009,4614.14,2026-09-12T07:13:00 +TX-00036,ACC-0019,ACC-0015,1342.28,2026-09-12T08:49:00 +TX-00012,ACC-0009,ACC-0014,16782.06,2026-09-12T16:45:00 +TX-00021,ACC-0016,ACC-0006,18163.40,2026-09-12T23:31:00 +TX-00020,ACC-0003,ACC-0007,10314.72,2026-09-13T02:15:00 +TX-00032,ACC-0019,ACC-0021,14762.61,2026-09-13T04:29:00 +TX-00035,ACC-0008,ACC-0009,12375.58,2026-09-14T19:56:00 +TX-00047,ACC-0013,ACC-0010,24356.62,2026-09-15T00:01:00 +TX-00019,ACC-0006,ACC-0011,17176.78,2026-09-15T00:28:00 +TX-00051,ACC-0018,ACC-0008,1226.38,2026-09-15T00:44:00 +TX-00026,ACC-0020,ACC-0006,24295.55,2026-09-15T11:58:00 +TX-00004,ACC-0001,ACC-0008,17736.04,2026-09-15T14:08:00 +TX-00005,ACC-0002,ACC-0001,3809.13,2026-09-16T04:33:00 +TX-00050,ACC-0016,ACC-0020,22660.49,2026-09-16T06:04:00 +TX-00059,ACC-0021,ACC-0006,17425.31,2026-09-16T22:17:00 +TX-00011,ACC-0016,ACC-0005,7413.98,2026-09-17T22:37:00 +TX-00037,ACC-0001,ACC-0012,13238.87,2026-09-18T00:30:00 +TX-00044,ACC-0014,ACC-0021,7584.47,2026-09-18T19:27:00 +TX-00056,ACC-0022,ACC-0006,916.03,2026-09-18T23:14:00 +TX-00031,ACC-0016,ACC-0023,18584.87,2026-09-19T02:11:00 +TX-00018,ACC-0001,ACC-0019,16750.03,2026-09-20T01:56:00 +TX-00058,ACC-0006,ACC-0024,9366.96,2026-09-21T07:53:00 +TX-00045,ACC-0004,ACC-0013,19902.21,2026-09-21T13:27:00 +TX-00010,ACC-0009,ACC-0006,5539.39,2026-09-21T15:06:00 +TX-00048,ACC-0017,ACC-0014,7880.60,2026-09-22T06:25:00 +TX-00049,ACC-0013,ACC-0021,8251.42,2026-09-22T18:49:00 +TX-00006,ACC-0014,ACC-0021,22073.56,2026-09-23T06:19:00 +TX-00060,ACC-0014,ACC-0002,1211.54,2026-09-23T18:29:00 +TX-00033,ACC-0002,ACC-0001,1347.28,2026-09-24T10:34:00 +TX-00029,ACC-0024,ACC-0019,6958.20,2026-09-25T08:07:00 +TX-00027,ACC-0022,ACC-0020,16655.47,2026-09-25T16:40:00 +TX-00028,ACC-0012,ACC-0006,8840.92,2026-09-25T17:08:00 +TX-00022,ACC-0004,ACC-0008,11271.48,2026-09-26T04:17:00 +TX-00043,ACC-0008,ACC-0005,16964.93,2026-09-26T06:47:00 +TX-00054,ACC-0013,ACC-0017,17249.54,2026-09-26T22:23:00 +TX-00057,ACC-0006,ACC-0023,14257.22,2026-09-27T03:01:00 +TX-00030,ACC-0005,ACC-0021,144.26,2026-09-27T05:24:00 +TX-00009,ACC-0014,ACC-0008,18448.21,2026-09-27T12:52:00 +TX-00040,ACC-0007,ACC-0018,16884.95,2026-09-27T16:07:00 +TX-00002,ACC-0019,ACC-0021,10370.35,2026-09-28T17:08:00 +TX-00025,ACC-0019,ACC-0020,17488.19,2026-09-29T00:40:00 +TX-00041,ACC-0008,ACC-0015,12644.28,2026-09-29T04:32:00 +TX-00017,ACC-0009,ACC-0023,15700.25,2026-09-29T13:11:00 +TX-00015,ACC-0011,ACC-0023,10958.45,2026-09-29T15:37:00 +TX-00042,ACC-0010,ACC-0022,20761.68,2026-09-29T23:49:00 +TX-00039,ACC-0010,ACC-0014,4582.28,2026-09-30T01:11:00 +TX-00052,ACC-0005,ACC-0013,2075.53,2026-10-01T05:16:00 diff --git a/examples/orchestration/samyama_airflow_dag.py b/examples/orchestration/samyama_airflow_dag.py new file mode 100644 index 00000000..97d9c86b --- /dev/null +++ b/examples/orchestration/samyama_airflow_dag.py @@ -0,0 +1,131 @@ +"""Reference Airflow DAG: load accounts and transfers into Samyama, then check them (INT-10). + + schema -> accounts -> transfers -> validate + +Copy this file and `samyama_pipeline.py` into your DAGs folder. Airflow puts +that folder on `sys.path`, which is how the import below resolves. + +Where the settings come from -- nothing about a deployment is written here: + +- The server and its credential come from an Airflow Connection, by default + `samyama_default`: host, port, `schema` (`http` or `https`, the convention + Airflow's own HTTP hook uses) and, if the server has an auth file, the token + in `password`. Optional `timeout_seconds` in the extras. +- The source files and batch size are DAG params, so a run can be pointed at + other files without editing the DAG. They default to the bundled sample. + +Retries: a step that fails because the server was unreachable is retried, with +backoff. A step that fails because the server refused the statement or the +credential, or because the input is wrong, fails at once -- retrying would read +the same files and get the same answer. See docs/ORCHESTRATION-AIRFLOW.md. + +Tested against Airflow 3.3 by `tests/test_airflow.py`. +""" + +from __future__ import annotations + +import functools +from datetime import timedelta +from urllib.parse import urlsplit + +from airflow.sdk import BaseHook, Param, dag, get_current_context, task +from airflow.sdk.exceptions import AirflowFailException + +import samyama_pipeline as sp + +DEFAULT_CONN_ID = "samyama_default" + + +def client_from_connection(conn_id: str) -> sp.SamyamaHttp: + conn = BaseHook.get_connection(conn_id) + host = conn.host or "" + if not host: + raise AirflowFailException(f"connection {conn_id!r} has no host") + base = host if "://" in host else f"{conn.schema or 'http'}://{host}" + if conn.port and urlsplit(base).port is None: + base = f"{base}:{conn.port}" + timeout = float(conn.extra_dejson.get("timeout_seconds", 30)) + return sp.SamyamaHttp(base, token=conn.password or None, timeout_seconds=timeout) + + +def fail_fast_on_permanent_errors(fn): + """Turn a failure that a retry cannot fix into one Airflow does not retry.""" + + @functools.wraps(fn) + def wrapper(*args, **kwargs): + try: + return fn(*args, **kwargs) + except (sp.SamyamaError, sp.SourceDataError) as e: + if sp.is_retryable(e): + raise + raise AirflowFailException(str(e)) from e + + return wrapper + + +def _settings() -> tuple[sp.SamyamaHttp, dict]: + params = get_current_context()["params"] + return client_from_connection(params["samyama_conn_id"]), params + + +@dag( + dag_id="samyama_accounts_transfers", + description="Load accounts and transfers from CSV into Samyama, then check the graph matches the files", + schedule=None, + catchup=False, + max_active_runs=1, + tags=["samyama", "reference"], + default_args={ + "retries": 3, + "retry_delay": timedelta(seconds=30), + "retry_exponential_backoff": True, + }, + params={ + "samyama_conn_id": Param(DEFAULT_CONN_ID, type="string", + description="Airflow Connection holding the server URL and token"), + "accounts_csv": Param(str(sp.SAMPLE_DIR / "accounts.csv"), type="string", + description="CSV with columns id, holder, country, opened"), + "transfers_csv": Param(str(sp.SAMPLE_DIR / "transfers.csv"), type="string", + description="CSV with columns id, src, dst, amount, at"), + "batch_size": Param(500, type="integer", minimum=1, + description="Rows per statement"), + }, +) +def samyama_accounts_transfers(): + @task + @fail_fast_on_permanent_errors + def schema() -> int: + client, _ = _settings() + return sp.ensure_schema(client) + + @task + @fail_fast_on_permanent_errors + def accounts() -> int: + client, params = _settings() + rows = sp.read_accounts(params["accounts_csv"]) + return sp.load_accounts(client, rows, params["batch_size"]) + + @task + @fail_fast_on_permanent_errors + def transfers() -> int: + client, params = _settings() + rows = sp.read_transfers(params["transfers_csv"]) + return sp.load_transfers(client, rows, params["batch_size"]) + + @task + @fail_fast_on_permanent_errors + def validate() -> dict: + client, params = _settings() + checks = sp.check_accounts(client, sp.read_accounts(params["accounts_csv"])) + checks += sp.check_transfers(client, sp.read_transfers(params["transfers_csv"])) + failed = sp.failures(checks) + if failed: + # The graph disagrees with the files it was loaded from. Retrying + # re-checks the same graph against the same files. + raise AirflowFailException("; ".join(failed)) + return {c.name: c.found for c in checks} + + schema() >> accounts() >> transfers() >> validate() + + +samyama_accounts_transfers() diff --git a/examples/orchestration/samyama_dagster.py b/examples/orchestration/samyama_dagster.py new file mode 100644 index 00000000..a0f2baf5 --- /dev/null +++ b/examples/orchestration/samyama_dagster.py @@ -0,0 +1,140 @@ +"""Reference Dagster assets: load accounts and transfers into Samyama, with checks (INT-10). + + samyama_schema -> accounts -> transfers + (check) (check) + +Load it as a code location with this directory on the path: + + dagster dev -f samyama_dagster.py + +Where the settings come from -- nothing about a deployment is written here: + +- `SAMYAMA_URL` (required) names the server's HTTP endpoint. `SAMYAMA_TOKEN` + is the bearer token, needed only when the server runs with an auth file. Both + are read through `dg.EnvVar`, so the secret is resolved when a run starts and + never stored with the code location. +- The source files and batch size are the `source_files` resource. They default + to the bundled sample; set them per run in the launchpad's resource config. + +Each asset's check is blocking: if the accounts in the graph do not match the +file, the transfers are not loaded on top of them. + +Retries: an unreachable server is retried with backoff. A refused statement or +credential, or wrong input, fails at once with `allow_retries=False` -- retrying +would read the same files and get the same answer. See +docs/ORCHESTRATION-DAGSTER.md. + +Tested against Dagster 1.13 by `tests/test_dagster.py`. +""" + +from __future__ import annotations + +import functools +import os +from typing import Optional + +import dagster as dg + +import samyama_pipeline as sp + + +class SamyamaResource(dg.ConfigurableResource): + """The server, its credential, and how long one statement may take.""" + + url: str + token: Optional[str] = None + timeout_seconds: float = 30.0 + + def client(self) -> sp.SamyamaHttp: + return sp.SamyamaHttp(self.url, token=self.token or None, timeout_seconds=self.timeout_seconds) + + +class SourceFiles(dg.ConfigurableResource): + """Where the rows come from, and how many go in one statement.""" + + accounts_csv: str = str(sp.SAMPLE_DIR / "accounts.csv") + transfers_csv: str = str(sp.SAMPLE_DIR / "transfers.csv") + batch_size: int = 500 + + +RETRY = dg.RetryPolicy(max_retries=3, delay=30, backoff=dg.Backoff.EXPONENTIAL) + + +def fail_fast_on_permanent_errors(fn): + """Turn a failure that a retry cannot fix into one Dagster does not retry.""" + + @functools.wraps(fn) + def wrapper(*args, **kwargs): + try: + return fn(*args, **kwargs) + except (sp.SamyamaError, sp.SourceDataError) as e: + if sp.is_retryable(e): + raise + raise dg.Failure(description=str(e), allow_retries=False) from e + + return wrapper + + +@dg.asset(retry_policy=RETRY, group_name="samyama") +@fail_fast_on_permanent_errors +def samyama_schema(samyama: SamyamaResource) -> dg.MaterializeResult: + return dg.MaterializeResult(metadata={"statements": sp.ensure_schema(samyama.client())}) + + +@dg.asset(deps=[samyama_schema], retry_policy=RETRY, group_name="samyama") +@fail_fast_on_permanent_errors +def accounts(samyama: SamyamaResource, source_files: SourceFiles) -> dg.MaterializeResult: + rows = sp.read_accounts(source_files.accounts_csv) + written = sp.load_accounts(samyama.client(), rows, source_files.batch_size) + return dg.MaterializeResult(metadata={"rows": written, "source": source_files.accounts_csv}) + + +@dg.asset(deps=[accounts], retry_policy=RETRY, group_name="samyama") +@fail_fast_on_permanent_errors +def transfers(samyama: SamyamaResource, source_files: SourceFiles) -> dg.MaterializeResult: + rows = sp.read_transfers(source_files.transfers_csv) + written = sp.load_transfers(samyama.client(), rows, source_files.batch_size) + return dg.MaterializeResult(metadata={"rows": written, "source": source_files.transfers_csv}) + + +def _result(checks: list[sp.Check]) -> dg.AssetCheckResult: + failed = sp.failures(checks) + return dg.AssetCheckResult( + passed=not failed, + description="; ".join(failed) or "the graph matches the source file", + metadata={c.name: f"{c.found} of {c.expected}" for c in checks}, + ) + + +@dg.asset_check(asset=accounts, blocking=True) +@fail_fast_on_permanent_errors +def accounts_match_source(samyama: SamyamaResource, source_files: SourceFiles) -> dg.AssetCheckResult: + return _result(sp.check_accounts(samyama.client(), sp.read_accounts(source_files.accounts_csv))) + + +@dg.asset_check(asset=transfers, blocking=True) +@fail_fast_on_permanent_errors +def transfers_match_source(samyama: SamyamaResource, source_files: SourceFiles) -> dg.AssetCheckResult: + return _result(sp.check_transfers(samyama.client(), sp.read_transfers(source_files.transfers_csv))) + + +ASSETS = [samyama_schema, accounts, transfers] +CHECKS = [accounts_match_source, transfers_match_source] + +def build_definitions() -> dg.Definitions: + return dg.Definitions( + assets=ASSETS, + asset_checks=CHECKS, + resources={ + "samyama": SamyamaResource( + url=dg.EnvVar("SAMYAMA_URL"), + # A server without an auth file needs no token, and an EnvVar + # that is unset fails the run -- so it is attached only when set. + token=dg.EnvVar("SAMYAMA_TOKEN") if os.environ.get("SAMYAMA_TOKEN") else None, + ), + "source_files": SourceFiles(), + }, + ) + + +defs = build_definitions() diff --git a/examples/orchestration/samyama_pipeline.py b/examples/orchestration/samyama_pipeline.py new file mode 100644 index 00000000..6c40cf00 --- /dev/null +++ b/examples/orchestration/samyama_pipeline.py @@ -0,0 +1,359 @@ +"""The load both reference pipelines run, with no orchestrator in it (INT-10). + +`samyama_airflow_dag.py` and `samyama_dagster.py` are thin: they decide when a +step runs, where its settings come from, and what a failure means to their +scheduler. What a step *does* lives here, once, so the two pipelines cannot +drift into loading the same files two different ways. + +Three properties every orchestrated load needs, and how this module gets each: + +- **A retry is safe.** Every write is a `MERGE` keyed on the source row's id, + so running a step twice leaves the graph as running it once. An orchestrator + retries on its own initiative -- a worker restart, a timeout -- and a load + that duplicates on retry corrupts the graph the first time that happens. +- **Values are bound, never interpolated.** Rows travel in the `params` of + `POST /api/query` (#1463), so a holder name containing a quote, or text that + looks like Cypher, is stored as the string it is. +- **A failure says whether retrying can help.** `SamyamaUnavailable` (no + connection, a timeout, a 5xx, a 408 or 429) might clear on its own; anything + else -- `SamyamaRejected` (the server refused the statement or the + credential), `SourceDataError` (the input is wrong), a load the server + reported inconsistently -- will not, and a scheduler that retries it only + delays the page to a human. `is_retryable` is the one place that decides. + +Standard library only: the module runs inside any orchestrator's worker +without adding a dependency to it. +""" + +from __future__ import annotations + +import csv +import json +import math +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterator, Optional, Sequence + +#: The sample files the pipelines read when they are not told otherwise. +SAMPLE_DIR = Path(__file__).resolve().parent / "sample" + + +class SamyamaError(Exception): + """A failure talking to the server. Retrying will not help unless it is + the `SamyamaUnavailable` subclass.""" + + +class SamyamaUnavailable(SamyamaError): + """The server could not be reached, failed, or asked the client to come + back later; a retry may succeed.""" + + +#: Client errors that say "not now" rather than "not this": a request timeout, +#: and a quota or rate limit. +RETRYABLE_4XX = frozenset({408, 429}) + + +def is_retryable(error: BaseException) -> bool: + """Whether a step that raised `error` is worth running again. + + Only an unavailable server is. A refused statement or credential, wrong + input, or a load the server reported inconsistently gets the same answer + the next time. + """ + return isinstance(error, SamyamaUnavailable) + + +class SamyamaRejected(SamyamaError): + """The server refused the request; a retry will get the same answer.""" + + def __init__(self, status: int, message: str): + super().__init__(f"HTTP {status}: {message}") + self.status = status + self.message = message + + +class SourceDataError(ValueError): + """The input files are wrong; a retry will read the same files.""" + + +@dataclass +class SamyamaHttp: + """A minimal client for `POST /api/query`. + + The Python SDK's `query` takes no parameters, so a pipeline built on it + would have to splice values into the statement text. This posts the + statement and its parameters separately, which is the property a loader + that reads untrusted files cannot do without. + """ + + url: str + token: Optional[str] = None + timeout_seconds: float = 30.0 + + def __post_init__(self) -> None: + if not self.url: + raise ValueError("a Samyama URL is required, e.g. http://localhost:8080") + self.url = self.url.rstrip("/") + + def query(self, cypher: str, params: Optional[dict] = None, read_only: bool = False) -> dict: + body: dict[str, Any] = {"query": cypher} + if params: + body["params"] = params + if read_only: + # The server refuses the statement if it writes (#1628). + body["read_only"] = True + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + request = urllib.request.Request( + f"{self.url}/api/query", + data=json.dumps(body).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response: + return json.loads(response.read()) + except urllib.error.HTTPError as e: + message = _error_message(e) + if e.code >= 500 or e.code in RETRYABLE_4XX: + # 503 is the server refusing writes after a failed flush (#1274): + # it can clear, so it is retryable along with every other 5xx. + raise SamyamaUnavailable(f"HTTP {e.code}: {message}") from e + raise SamyamaRejected(e.code, message) from e + except (urllib.error.URLError, TimeoutError, ConnectionError) as e: + raise SamyamaUnavailable(f"{self.url}: {e}") from e + + def scalar(self, cypher: str, params: Optional[dict] = None) -> Any: + """The single value a one-row, one-column read returns.""" + records = self.query(cypher, params, read_only=True)["records"] + if len(records) != 1 or len(records[0]) != 1: + raise SamyamaError(f"expected one value, got {records!r} for: {cypher}") + return records[0][0] + + +def _error_message(e: urllib.error.HTTPError) -> str: + raw = e.read().decode("utf-8", "replace") + try: + return json.loads(raw).get("error", raw) + except (ValueError, AttributeError): + return raw or e.reason + + +# -------------------------------------------------------------------------- +# Reading the source files +# -------------------------------------------------------------------------- + +ACCOUNT_COLUMNS = ("id", "holder", "country", "opened") +TRANSFER_COLUMNS = ("id", "src", "dst", "amount", "at") + + +def read_accounts(path: str | Path) -> list[dict]: + rows = _read_csv(path, ACCOUNT_COLUMNS) + _reject_duplicate_ids(path, rows) + return [{"id": r["id"], "holder": r["holder"], "country": r["country"], "opened": r["opened"]} + for _, r in rows] + + +def read_transfers(path: str | Path) -> list[dict]: + rows = _read_csv(path, TRANSFER_COLUMNS) + _reject_duplicate_ids(path, rows) + out = [] + for line, r in rows: + try: + amount = float(r["amount"]) + except ValueError: + raise SourceDataError(f"{path}:{line}: amount {r['amount']!r} is not a number") from None + if not math.isfinite(amount): + raise SourceDataError(f"{path}:{line}: amount {r['amount']!r} is not finite") + out.append({"id": r["id"], "src": r["src"], "dst": r["dst"], "amount": amount, "at": r["at"]}) + return out + + +def _read_csv(path: str | Path, columns: Sequence[str]) -> list[tuple[int, dict]]: + path = Path(path) + if not path.is_file(): + raise SourceDataError(f"{path}: no such file") + with path.open(newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + missing = [c for c in columns if c not in (reader.fieldnames or [])] + if missing: + raise SourceDataError(f"{path}: missing column(s) {', '.join(missing)}") + rows = [] + # Line 1 is the header, so the first data row is line 2. + for line, row in enumerate(reader, start=2): + for c in columns: + if row.get(c) is None or row[c].strip() == "": + raise SourceDataError(f"{path}:{line}: empty {c!r}") + rows.append((line, row)) + return rows + + +def _reject_duplicate_ids(path: str | Path, rows: list[tuple[int, dict]]) -> None: + # Two rows with one id would MERGE into one node or edge, and the second + # row's values would silently win. Which one is right is a question for + # whoever owns the file, not for the loader. + first_seen: dict[str, int] = {} + for line, r in rows: + if r["id"] in first_seen: + raise SourceDataError(f"{path}:{line}: id {r['id']!r} already on line {first_seen[r['id']]}") + first_seen[r["id"]] = line + + +def batches(rows: Sequence[dict], size: int) -> Iterator[list[dict]]: + if size < 1: + raise ValueError(f"batch size must be at least 1, got {size}") + for start in range(0, len(rows), size): + yield list(rows[start:start + size]) + + +# -------------------------------------------------------------------------- +# The steps +# -------------------------------------------------------------------------- + +#: Run before any load. A unique constraint rather than a plain index: it backs +#: the `{id: ...}` lookups below with an index the same way, and it also makes +#: the engine refuse a second `:Account` with an id that is already taken, so a +#: writer that uses CREATE where it should MERGE fails instead of duplicating. +#: `IF NOT EXISTS` makes the step safe to repeat. Over a graph that already +#: holds a duplicate id the statement is refused, naming the value. +SCHEMA_STATEMENTS = ( + "CREATE CONSTRAINT IF NOT EXISTS FOR (a:Account) REQUIRE a.id IS UNIQUE", +) + +UPSERT_ACCOUNTS = """ +UNWIND $rows AS r +MERGE (a:Account {id: r.id}) +SET a.holder = r.holder, a.country = r.country, a.opened = r.opened +RETURN count(a) AS written +""" + +FOUND_ACCOUNTS = """ +UNWIND $ids AS id +MATCH (a:Account {id: id}) +RETURN collect(a.id) AS found +""" + +# Both endpoints in ONE comma-separated MATCH. Written as two MATCH clauses -- +# which means the same thing -- the planner does not look each one up by id: it +# builds every pair of accounts and then filters, so the cost grows with the +# square of the number of accounts in the graph rather than with the rows. +UPSERT_TRANSFERS = """ +UNWIND $rows AS r +MATCH (s:Account {id: r.src}), (d:Account {id: r.dst}) +MERGE (s)-[t:TRANSFER {id: r.id}]->(d) +SET t.amount = r.amount, t.at = r.at +RETURN count(t) AS written +""" + +#: Each account, found by its id, holding the values the file holds. +CHECK_ACCOUNTS = """ +UNWIND $rows AS r +MATCH (a:Account {id: r.id}) +WHERE a.holder = r.holder AND a.country = r.country AND a.opened = r.opened +RETURN count(a) AS matched +""" + +# Each transfer, between the accounts the file names, holding its values. The +# endpoints are named variables: the engine refuses a property map that is not +# a literal on an anonymous node (`(:Account {id: r.src})`). +CHECK_TRANSFERS = """ +UNWIND $rows AS r +MATCH (s:Account {id: r.src})-[t:TRANSFER]->(d:Account {id: r.dst}) +WHERE t.id = r.id AND t.amount = r.amount AND t.at = r.at +RETURN count(t) AS matched +""" + +# Edge ids have no index to look them up by, so this is one pass over the +# TRANSFER edges, grouped by id -- linear in the edges, where asking id by id +# would be linear in the edges once per id. +DUPLICATED_TRANSFER_IDS = """ +MATCH ()-[t:TRANSFER]->() +WITH t.id AS id, count(t) AS n +WHERE n > 1 +RETURN collect(id) AS duplicated +""" + + +def ensure_schema(client: SamyamaHttp) -> int: + for statement in SCHEMA_STATEMENTS: + client.query(statement) + return len(SCHEMA_STATEMENTS) + + +def load_accounts(client: SamyamaHttp, rows: Sequence[dict], batch_size: int) -> int: + written = 0 + for batch in batches(rows, batch_size): + written += client.query(UPSERT_ACCOUNTS, {"rows": batch})["records"][0][0] + if written != len(rows): + raise SamyamaError(f"wrote {written} accounts for {len(rows)} rows") + return written + + +def load_transfers(client: SamyamaHttp, rows: Sequence[dict], batch_size: int) -> int: + # Checked before anything is written. A transfer whose account does not + # exist would otherwise match nothing and be dropped without a word: the + # MATCH finds no row, so the MERGE never runs and the statement succeeds. + endpoints = sorted({r["src"] for r in rows} | {r["dst"] for r in rows}) + found = set(client.scalar(FOUND_ACCOUNTS, {"ids": endpoints})) if endpoints else set() + missing = [i for i in endpoints if i not in found] + if missing: + raise SourceDataError( + f"{len(missing)} account(s) referenced by transfers do not exist: " + + ", ".join(sorted(missing)[:20]) + + (" ..." if len(missing) > 20 else "") + ) + written = 0 + for batch in batches(rows, batch_size): + written += client.query(UPSERT_TRANSFERS, {"rows": batch})["records"][0][0] + if written != len(rows): + raise SamyamaError(f"wrote {written} transfers for {len(rows)} rows") + return written + + +@dataclass +class Check: + name: str + expected: int + found: int + detail: str = field(default="") + + @property + def passed(self) -> bool: + return self.found == self.expected + + +def check_accounts(client: SamyamaHttp, rows: Sequence[dict]) -> list[Check]: + ids = [r["id"] for r in rows] + return [ + Check("accounts present with the file's values", len(rows), + client.scalar(CHECK_ACCOUNTS, {"rows": list(rows)})), + # A second node with the same id -- written by something that used + # CREATE rather than MERGE -- passes the check above and fails this one. + Check("no account id held by more than one node", len(set(ids)), + client.scalar("UNWIND $ids AS id MATCH (a:Account {id: id}) RETURN count(a)", + {"ids": ids})), + ] + + +def check_transfers(client: SamyamaHttp, rows: Sequence[dict]) -> list[Check]: + ids = {r["id"] for r in rows} + # Only this file's ids count against it: a duplicate some other load wrote + # is that load's to answer for. + duplicated = sorted(ids.intersection(client.scalar(DUPLICATED_TRANSFER_IDS))) + return [ + Check("transfers present between the file's accounts with its values", len(rows), + client.scalar(CHECK_TRANSFERS, {"rows": list(rows)})), + # MERGE matches on both endpoints and the id, so a source row whose + # account changed between two runs adds a second edge with the old + # id rather than moving the first. This is what sees it. + Check("no transfer id held by more than one edge", 0, len(duplicated), + ", ".join(duplicated[:20])), + ] + + +def failures(checks: Sequence[Check]) -> list[str]: + return [f"{c.name}: expected {c.expected}, found {c.found}" + (f" ({c.detail})" if c.detail else "") + for c in checks if not c.passed] diff --git a/examples/orchestration/tests/conftest.py b/examples/orchestration/tests/conftest.py new file mode 100644 index 00000000..f1781584 --- /dev/null +++ b/examples/orchestration/tests/conftest.py @@ -0,0 +1,199 @@ +"""A real Samyama server for each test, on spare ports, in a throwaway directory. + +The pipelines are tested against the engine, not a mock of it: what INT-10 +asks for is an integration that has been run, and the defects worth catching +here -- a statement the engine refuses, a MATCH that silently finds nothing, a +parameter that arrives as the wrong type -- are the ones a mock would agree +with. + +The binary is `$SAMYAMA_BIN`, or the repository's release build if that exists. +Neither present is an error, not a skip: a suite that skips when its server is +missing reports green while testing nothing. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +import pytest + +HERE = Path(__file__).resolve().parent +PIPELINE_DIR = HERE.parent +REPO_ROOT = PIPELINE_DIR.parent.parent + +sys.path.insert(0, str(PIPELINE_DIR)) + +STARTUP_SECONDS = float(os.environ.get("SAMYAMA_STARTUP_SECONDS", "60")) + + +def samyama_bin() -> Path: + configured = os.environ.get("SAMYAMA_BIN") + path = Path(configured) if configured else REPO_ROOT / "target" / "release" / "samyama" + if not path.is_file() or not os.access(path, os.X_OK): + pytest.fail( + f"no Samyama server binary at {path}. Build one with " + "`cargo build --release --bin samyama`, or set SAMYAMA_BIN.", + pytrace=False, + ) + return path + + +def free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@dataclass +class Engine: + url: str + token: Optional[str] + log: Path + + def query(self, cypher: str, params: Optional[dict] = None) -> list: + body = {"query": cypher, "params": params or {}} + headers = {"Content-Type": "application/json"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + req = urllib.request.Request(f"{self.url}/api/query", json.dumps(body).encode(), headers) + try: + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read())["records"] + except urllib.error.HTTPError as e: + raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')}") from None + + def count(self, cypher: str, params: Optional[dict] = None) -> int: + return self.query(cypher, params)[0][0] + + +def _start(tmp_path: Path, auth_line: Optional[str] = None): + binary = samyama_bin() + data = tmp_path / "samyama-data" + data.mkdir() + http_port, resp_port = free_port(), free_port() + args = [str(binary), "--host", "127.0.0.1", "--port", str(resp_port), + "--http-port", str(http_port), "--data-path", str(data)] + if auth_line: + auth_file = tmp_path / "credentials" + auth_file.write_text(auth_line + "\n") + args += ["--auth-file", str(auth_file)] + log = tmp_path / "samyama.log" + # cwd is the scratch directory: the server resolves some defaults against + # its working directory, and the repository is not the place for them. + proc = subprocess.Popen(args, cwd=tmp_path, stdout=log.open("w"), stderr=subprocess.STDOUT) + url = f"http://127.0.0.1:{http_port}" + deadline = time.monotonic() + STARTUP_SECONDS + while True: + if proc.poll() is not None: + pytest.fail(f"samyama exited with {proc.returncode}:\n{log.read_text()[-4000:]}", pytrace=False) + try: + urllib.request.urlopen(f"{url}/api/status", timeout=2).close() + break + except urllib.error.HTTPError: + break # answering, even if only to say 401 + except (urllib.error.URLError, ConnectionError, TimeoutError): + if time.monotonic() > deadline: + proc.kill() + pytest.fail(f"samyama did not answer on {url} within {STARTUP_SECONDS:.0f}s:\n" + f"{log.read_text()[-4000:]}", pytrace=False) + time.sleep(0.2) + return proc, url, log + + +def _stop(proc: subprocess.Popen) -> None: + proc.terminate() + try: + proc.wait(timeout=20) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + +@pytest.fixture +def engine(tmp_path): + proc, url, log = _start(tmp_path) + try: + yield Engine(url, None, log) + finally: + _stop(proc) + + +@pytest.fixture +def secured_engine(tmp_path): + """A server that refuses requests without a token, and the token it accepts.""" + # The server's own subcommand writes the credential, so the test does not + # restate how a token is hashed. It prints the file line, then the token. + out = subprocess.run([str(samyama_bin()), "auth-token", "pipeline"], + capture_output=True, text=True, check=True, cwd=tmp_path).stdout + lines = [line for line in out.splitlines() if line and not line.startswith("#")] + credential, token = lines[0], lines[1] + # Write, not admin: a pipeline that loads data has no business managing tenants. + proc, url, log = _start(tmp_path, auth_line=f"{credential}:roles=write") + try: + yield Engine(url, token, log) + finally: + _stop(proc) + + +# --------------------------------------------------------------------------- +# Source files, and what the graph should hold after loading them +# --------------------------------------------------------------------------- + +import samyama_pipeline as sp # noqa: E402 (needs the sys.path entry above) + + +@dataclass +class Source: + accounts_csv: Path + transfers_csv: Path + + @property + def accounts(self) -> list[dict]: + return sp.read_accounts(self.accounts_csv) + + @property + def transfers(self) -> list[dict]: + return sp.read_transfers(self.transfers_csv) + + +@pytest.fixture +def sample() -> Source: + return Source(sp.SAMPLE_DIR / "accounts.csv", sp.SAMPLE_DIR / "transfers.csv") + + +def assert_graph_matches(engine: Engine, source: Source) -> None: + """The graph holds exactly the source files, read back with plain Cypher. + + Deliberately not `samyama_pipeline.check_*`: the pipelines validate with + those, and a test that asked them whether they were right would pass + whenever they were wrong in the same way twice. + """ + accounts, transfers = source.accounts, source.transfers + assert engine.count("MATCH (a:Account) RETURN count(a)") == len(accounts) + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == len(transfers) + + got_accounts = { + r[0]: {"holder": r[1], "country": r[2], "opened": r[3]} + for r in engine.query("MATCH (a:Account) RETURN a.id, a.holder, a.country, a.opened") + } + assert got_accounts == { + a["id"]: {"holder": a["holder"], "country": a["country"], "opened": a["opened"]} + for a in accounts + } + + got_transfers = { + r[0]: (r[1], r[2], r[3], r[4]) + for r in engine.query( + "MATCH (s:Account)-[t:TRANSFER]->(d:Account) RETURN t.id, s.id, d.id, t.amount, t.at") + } + assert got_transfers == {t["id"]: (t["src"], t["dst"], t["amount"], t["at"]) for t in transfers} diff --git a/examples/orchestration/tests/test_airflow.py b/examples/orchestration/tests/test_airflow.py new file mode 100644 index 00000000..b5af348a --- /dev/null +++ b/examples/orchestration/tests/test_airflow.py @@ -0,0 +1,111 @@ +"""The reference Airflow DAG, run by Airflow against a real server. + +`dag.test()` runs every task in this process through Airflow's own task +runner: params are resolved, the Connection is looked up through Airflow's +secrets backends, and a task's failure is classified the way a scheduler would +classify it. It needs a metadata database, so the session gets a throwaway one. + +Airflow reads its configuration when it is first imported, so nothing here +imports it at module level: the environment is set first, in the fixture. +""" + +from __future__ import annotations + +import csv +import importlib +import json +import os +import subprocess +import sys + +import pytest + +from conftest import PIPELINE_DIR, Source, assert_graph_matches + +DAG_ID = "samyama_accounts_transfers" + + +@pytest.fixture(scope="session") +def airflow_dag(tmp_path_factory): + home = tmp_path_factory.mktemp("airflow-home") + env = { + "AIRFLOW_HOME": str(home), + "AIRFLOW__CORE__LOAD_EXAMPLES": "False", + "AIRFLOW__CORE__DAGS_FOLDER": str(PIPELINE_DIR), + } + os.environ.update(env) + migrate = subprocess.run([sys.executable, "-m", "airflow", "db", "migrate"], + capture_output=True, text=True, env={**os.environ, **env}) + assert migrate.returncode == 0, migrate.stdout[-3000:] + migrate.stderr[-3000:] + module = importlib.import_module("samyama_airflow_dag") + return module.samyama_accounts_transfers() + + +def connect(monkeypatch, engine, conn_id="samyama_default"): + """Point an Airflow Connection at the server, the way an operator would.""" + host_port = engine.url.split("://", 1)[1] + host, port = host_port.rsplit(":", 1) + conn = {"conn_type": "http", "host": host, "port": int(port), "schema": "http"} + if engine.token: + conn["password"] = engine.token + monkeypatch.setenv(f"AIRFLOW_CONN_{conn_id.upper()}", json.dumps(conn)) + + +def run(dag, source: Source, **conf): + run_conf = {"accounts_csv": str(source.accounts_csv), + "transfers_csv": str(source.transfers_csv), **conf} + dr = dag.test(run_conf=run_conf) + states = {ti.task_id: (ti.state, ti.try_number) for ti in dr.get_task_instances()} + return dr.state, states + + +def test_the_dag_loads_the_sample_and_validates_it(airflow_dag, engine, sample, monkeypatch): + connect(monkeypatch, engine) + state, tasks = run(airflow_dag, sample, batch_size=7) + assert state == "success", tasks + assert {t: s for t, (s, _) in tasks.items()} == { + "schema": "success", "accounts": "success", "transfers": "success", "validate": "success"} + assert_graph_matches(engine, sample) + + +def test_running_the_dag_twice_leaves_the_graph_as_once(airflow_dag, engine, sample, monkeypatch): + connect(monkeypatch, engine) + assert run(airflow_dag, sample, batch_size=7)[0] == "success" + assert run(airflow_dag, sample, batch_size=13)[0] == "success" + assert_graph_matches(engine, sample) + + +def test_bad_input_fails_the_run_without_retrying(airflow_dag, engine, sample, monkeypatch, tmp_path): + connect(monkeypatch, engine) + bad = tmp_path / "transfers.csv" + with bad.open("w", newline="") as f: + w = csv.writer(f, lineterminator="\n") + w.writerow(["id", "src", "dst", "amount", "at"]) + w.writerow(["TX-1", sample.accounts[0]["id"], "ACC-NOWHERE", "1.00", "2026-01-01T00:00:00"]) + state, tasks = run(airflow_dag, Source(sample.accounts_csv, bad)) + assert state == "failed" + assert tasks["transfers"] == ("failed", 1), tasks # one try: the DAG allows three retries + assert tasks["validate"][0] == "upstream_failed" + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +def test_the_token_comes_from_the_connection(airflow_dag, secured_engine, sample, monkeypatch): + connect(monkeypatch, secured_engine) + state, tasks = run(airflow_dag, sample) + assert state == "success", tasks + assert_graph_matches(secured_engine, sample) + + +def test_a_connection_without_the_token_fails_without_retrying(airflow_dag, secured_engine, sample, + monkeypatch): + connect(monkeypatch, secured_engine.__class__(secured_engine.url, None, secured_engine.log)) + state, tasks = run(airflow_dag, sample) + assert state == "failed" + assert tasks["schema"] == ("failed", 1), tasks + + +def test_the_connection_id_is_a_param(airflow_dag, engine, sample, monkeypatch): + connect(monkeypatch, engine, conn_id="graph_prod_like") + state, tasks = run(airflow_dag, sample, samyama_conn_id="graph_prod_like") + assert state == "success", tasks + assert_graph_matches(engine, sample) diff --git a/examples/orchestration/tests/test_dagster.py b/examples/orchestration/tests/test_dagster.py new file mode 100644 index 00000000..a1665322 --- /dev/null +++ b/examples/orchestration/tests/test_dagster.py @@ -0,0 +1,117 @@ +"""The reference Dagster assets and checks, run by Dagster against a real server. + +`materialize` executes the assets and their checks in this process through +Dagster's own executor, with the resources an operator would configure. +""" + +from __future__ import annotations + +import csv + +import dagster as dg + +import samyama_dagster as sd +from conftest import Source, assert_graph_matches + + +def resources(engine, source: Source, batch_size: int = 500, token=None) -> dict: + return { + "samyama": sd.SamyamaResource(url=engine.url, token=token), + "source_files": sd.SourceFiles(accounts_csv=str(source.accounts_csv), + transfers_csv=str(source.transfers_csv), + batch_size=batch_size), + } + + +def run(res: dict) -> dg.ExecuteInProcessResult: + return dg.materialize([*sd.ASSETS, *sd.CHECKS], resources=res, raise_on_error=False) + + +def checks(result: dg.ExecuteInProcessResult) -> dict: + return {e.check_name: e.passed for e in result.get_asset_check_evaluations()} + + +def materialized(result: dg.ExecuteInProcessResult) -> set: + return {e.asset_key.to_user_string() for e in result.get_asset_materialization_events()} + + +def test_the_assets_load_the_sample_and_the_checks_pass(engine, sample): + result = run(resources(engine, sample, batch_size=7)) + assert result.success + assert materialized(result) == {"samyama_schema", "accounts", "transfers"} + assert checks(result) == {"accounts_match_source": True, "transfers_match_source": True} + assert_graph_matches(engine, sample) + + +def test_materializing_twice_leaves_the_graph_as_once(engine, sample): + assert run(resources(engine, sample, batch_size=7)).success + assert run(resources(engine, sample, batch_size=13)).success + assert_graph_matches(engine, sample) + + +def test_bad_input_fails_the_asset_without_retrying(engine, sample, tmp_path): + bad = tmp_path / "transfers.csv" + with bad.open("w", newline="") as f: + w = csv.writer(f, lineterminator="\n") + w.writerow(["id", "src", "dst", "amount", "at"]) + w.writerow(["TX-1", sample.accounts[0]["id"], "ACC-NOWHERE", "1.00", "2026-01-01T00:00:00"]) + result = run(resources(engine, Source(sample.accounts_csv, bad))) + assert not result.success + assert "transfers" not in materialized(result) + failures = [e for e in result.all_events if e.event_type_value == "STEP_FAILURE"] + assert [e.step_key for e in failures] == ["transfers"] + assert "ACC-NOWHERE" in failures[0].event_specific_data.error.message + assert not [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +def test_a_failed_accounts_check_blocks_the_transfers(engine, sample, monkeypatch): + # Make the graph disagree with the file between the load and its check: + # the check must fail, and being blocking, stop `transfers` from loading. + real = sd.sp.load_accounts + + def load_then_corrupt(client, rows, batch_size): + written = real(client, rows, batch_size) + client.query("MATCH (a:Account {id: $id}) SET a.holder = 'tampered'", {"id": rows[0]["id"]}) + return written + + monkeypatch.setattr(sd.sp, "load_accounts", load_then_corrupt) + result = run(resources(engine, sample)) + assert not result.success + assert checks(result) == {"accounts_match_source": False} + assert "transfers" not in materialized(result) + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +def test_the_token_comes_from_the_resource(secured_engine, sample): + result = run(resources(secured_engine, sample, token=secured_engine.token)) + assert result.success + assert_graph_matches(secured_engine, sample) + + +def test_a_resource_without_the_token_fails_without_retrying(secured_engine, sample): + result = run(resources(secured_engine, sample, token=None)) + assert not result.success + failures = [e for e in result.all_events if e.event_type_value == "STEP_FAILURE"] + assert [e.step_key for e in failures] == ["samyama_schema"] + assert "401" in failures[0].event_specific_data.error.message + assert not [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] + + +def test_the_definitions_read_the_server_from_the_environment(engine, sample, monkeypatch): + # The shipped definitions, not a test-built copy: the URL arrives through EnvVar. + monkeypatch.delenv("SAMYAMA_TOKEN", raising=False) + monkeypatch.setenv("SAMYAMA_URL", engine.url) + job = sd.build_definitions().resolve_implicit_global_asset_job_def() + result = job.execute_in_process(raise_on_error=False) + assert result.success + assert_graph_matches(engine, sample) + + +def test_the_definitions_read_the_token_from_the_environment(secured_engine, sample, monkeypatch): + monkeypatch.setenv("SAMYAMA_URL", secured_engine.url) + monkeypatch.setenv("SAMYAMA_TOKEN", secured_engine.token) + job = sd.build_definitions().resolve_implicit_global_asset_job_def() + result = job.execute_in_process(raise_on_error=False) + assert result.success + assert_graph_matches(secured_engine, sample) diff --git a/examples/orchestration/tests/test_docs.py b/examples/orchestration/tests/test_docs.py new file mode 100644 index 00000000..66931fb8 --- /dev/null +++ b/examples/orchestration/tests/test_docs.py @@ -0,0 +1,43 @@ +"""The Cypher the orchestration docs show is the Cypher the pipelines run. + +CH-DOC-EXEC checks that a documented statement parses. It cannot see that the +statement on the page is no longer the one in `samyama_pipeline.py` -- and a +reader copies the page, not the module. +""" + +from __future__ import annotations + +import re + +import pytest + +import samyama_pipeline as sp +from conftest import REPO_ROOT + +DOCS = ["ORCHESTRATION.md", "ORCHESTRATION-AIRFLOW.md", "ORCHESTRATION-DAGSTER.md"] + + +def normalise(text: str) -> str: + return " ".join(text.split()) + + +def pipeline_statements() -> set[str]: + return {normalise(v) for k, v in vars(sp).items() + if k.isupper() and isinstance(v, str) and ("MATCH" in v or "MERGE" in v)} | { + normalise(s) for s in sp.SCHEMA_STATEMENTS} + + +@pytest.mark.parametrize("doc", DOCS) +def test_every_cypher_block_is_a_statement_the_pipeline_runs(doc): + text = (REPO_ROOT / "docs" / doc).read_text(encoding="utf-8") + blocks = re.findall(r"```cypher\n(.*?)```", text, flags=re.S) + known = pipeline_statements() + for block in blocks: + assert normalise(block) in known, ( + f"docs/{doc} shows a statement samyama_pipeline.py does not run:\n{block}") + + +def test_the_overview_shows_at_least_one_statement(): + # Without this, deleting every block would pass the test above. + text = (REPO_ROOT / "docs" / "ORCHESTRATION.md").read_text(encoding="utf-8") + assert "```cypher" in text diff --git a/examples/orchestration/tests/test_pipeline.py b/examples/orchestration/tests/test_pipeline.py new file mode 100644 index 00000000..3ff38643 --- /dev/null +++ b/examples/orchestration/tests/test_pipeline.py @@ -0,0 +1,268 @@ +"""The shared load steps, against a real server and with no orchestrator.""" + +from __future__ import annotations + +import csv +from pathlib import Path + +import pytest + +import samyama_pipeline as sp +from conftest import Source, assert_graph_matches + + +def write_csv(path: Path, columns, rows) -> Path: + with path.open("w", newline="", encoding="utf-8") as f: + w = csv.writer(f, lineterminator="\n") + w.writerow(columns) + w.writerows(rows) + return path + + +def load_all(client, source: Source, batch_size: int) -> None: + sp.ensure_schema(client) + sp.load_accounts(client, source.accounts, batch_size) + sp.load_transfers(client, source.transfers, batch_size) + + +# --- reading ----------------------------------------------------------------- + +def test_a_duplicate_id_names_both_lines(tmp_path): + f = write_csv(tmp_path / "a.csv", sp.ACCOUNT_COLUMNS, + [["A1", "x", "IN", "2020-01-01"], ["A2", "y", "IN", "2020-01-01"], + ["A1", "z", "GB", "2021-01-01"]]) + with pytest.raises(sp.SourceDataError, match=r"a\.csv:4: id 'A1' already on line 2"): + sp.read_accounts(f) + + +def test_a_missing_column_is_named(tmp_path): + f = write_csv(tmp_path / "t.csv", ["id", "src", "dst", "at"], [["T1", "A1", "A2", "x"]]) + with pytest.raises(sp.SourceDataError, match="missing column.*amount"): + sp.read_transfers(f) + + +@pytest.mark.parametrize("amount", ["twelve", "nan", "inf"]) +def test_an_amount_that_is_not_a_finite_number_is_refused(tmp_path, amount): + f = write_csv(tmp_path / "t.csv", sp.TRANSFER_COLUMNS, [["T1", "A1", "A2", amount, "x"]]) + with pytest.raises(sp.SourceDataError, match=r"t\.csv:2: amount"): + sp.read_transfers(f) + + +def test_an_empty_field_is_refused(tmp_path): + f = write_csv(tmp_path / "a.csv", sp.ACCOUNT_COLUMNS, [["A1", " ", "IN", "2020-01-01"]]) + with pytest.raises(sp.SourceDataError, match="empty 'holder'"): + sp.read_accounts(f) + + +def test_batches_cover_every_row_once(): + rows = [{"id": i} for i in range(23)] + got = list(sp.batches(rows, 5)) + assert [len(b) for b in got] == [5, 5, 5, 5, 3] + assert [r for b in got for r in b] == rows + with pytest.raises(ValueError): + list(sp.batches(rows, 0)) + + +# --- loading ----------------------------------------------------------------- + +def test_the_sample_loads_exactly(engine, sample): + # A batch size that divides neither file, so the last batch of each is short. + load_all(sp.SamyamaHttp(engine.url), sample, batch_size=7) + assert_graph_matches(engine, sample) + + +def test_loading_twice_is_loading_once(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, batch_size=7) + load_all(client, sample, batch_size=11) + assert_graph_matches(engine, sample) + + +def test_values_are_stored_as_given_not_read_as_cypher(engine, tmp_path): + hostile = [ + "O'Brien", + 'say "hi"', + "x'}) DETACH DELETE (n) //", + "back\\slash", + "$rows", + "名前 — ünïcödé", + ] + accounts = write_csv(tmp_path / "a.csv", sp.ACCOUNT_COLUMNS, + [[f"A{i}", h, "IN", "2020-01-01"] for i, h in enumerate(hostile)]) + transfers = write_csv(tmp_path / "t.csv", sp.TRANSFER_COLUMNS, + [["T0", "A0", "A1", "1.5", "2026-01-01T00:00:00"]]) + source = Source(accounts, transfers) + load_all(sp.SamyamaHttp(engine.url), source, batch_size=2) + assert_graph_matches(engine, source) + + +def test_a_transfer_to_an_unknown_account_writes_nothing(engine, sample, tmp_path): + client = sp.SamyamaHttp(engine.url) + sp.ensure_schema(client) + sp.load_accounts(client, sample.accounts, 500) + + rows = [[t["id"], t["src"], t["dst"], t["amount"], t["at"]] for t in sample.transfers] + rows.append(["TX-ORPHAN", sample.accounts[0]["id"], "ACC-DOES-NOT-EXIST", "1.00", + "2026-01-01T00:00:00"]) + bad = Source(sample.accounts_csv, write_csv(tmp_path / "t.csv", sp.TRANSFER_COLUMNS, rows)) + + with pytest.raises(sp.SourceDataError, match="ACC-DOES-NOT-EXIST"): + sp.load_transfers(client, bad.transfers, 7) + # Checked before the first batch, so not even the valid rows went in. + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +# --- validating -------------------------------------------------------------- + +def test_the_checks_pass_on_a_correct_load(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + checks = sp.check_accounts(client, sample.accounts) + sp.check_transfers(client, sample.transfers) + assert sp.failures(checks) == [] + # Not vacuous: the checks that count what is present expected the files' + # rows, not zero of them. (The duplicate check expects zero by design.) + present = {c.name: c.expected for c in checks if "present" in c.name} + assert present == { + "accounts present with the file's values": len(sample.accounts), + "transfers present between the file's accounts with its values": len(sample.transfers), + } + + +def test_the_checks_see_a_changed_value(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + victim = sample.transfers[0] + engine.query("MATCH ()-[t:TRANSFER {id: $id}]->() SET t.amount = $a", + {"id": victim["id"], "a": victim["amount"] + 0.01}) + failed = sp.failures(sp.check_transfers(client, sample.transfers)) + assert len(failed) == 1 and "values" in failed[0], failed + + +def _create_account(engine, a): + engine.query("CREATE (:Account {id: $id, holder: $h, country: $c, opened: $o})", + {"id": a["id"], "h": a["holder"], "c": a["country"], "o": a["opened"]}) + + +def test_after_the_schema_step_the_engine_refuses_a_duplicate_account(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + with pytest.raises(Exception, match="Constraint violation"): + _create_account(engine, sample.accounts[0]) + assert engine.count("MATCH (a:Account) RETURN count(a)") == len(sample.accounts) + + +def test_the_schema_step_refuses_a_graph_that_already_holds_a_duplicate(engine, sample): + _create_account(engine, sample.accounts[0]) + _create_account(engine, sample.accounts[0]) + with pytest.raises(sp.SamyamaRejected, match="duplicate value"): + sp.ensure_schema(sp.SamyamaHttp(engine.url)) + + +def test_the_checks_see_a_duplicate_node_where_no_constraint_stops_it(engine, sample): + # A graph some other writer populated, without the schema step. + client = sp.SamyamaHttp(engine.url) + sp.load_accounts(client, sample.accounts, 500) + _create_account(engine, sample.accounts[0]) + failed = sp.failures(sp.check_accounts(client, sample.accounts)) + assert any("more than one node" in f for f in failed), failed + + +def test_the_checks_see_a_duplicate_edge(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + t = sample.transfers[0] + engine.query("MATCH (s:Account {id: $s}), (d:Account {id: $d}) " + "CREATE (s)-[:TRANSFER {id: $id, amount: $a, at: $at}]->(d)", + {"s": t["src"], "d": t["dst"], "id": t["id"], "a": t["amount"], "at": t["at"]}) + failed = sp.failures(sp.check_transfers(client, sample.transfers)) + # An exact copy -- same id, endpoints and values -- trips both checks: + # it is one row too many present, and its id is held twice. + n = len(sample.transfers) + assert failed == [ + f"transfers present between the file's accounts with its values: expected {n}, found {n + 1}", + f"no transfer id held by more than one edge: expected 0, found 1 ({t['id']})", + ] + + +def test_a_transfer_whose_account_changed_between_runs_is_reported(engine, sample, tmp_path): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + # The same transfer id, re-sent with a different destination. + moved = dict(sample.transfers[0]) + moved["dst"] = next(a["id"] for a in sample.accounts if a["id"] not in (moved["src"], moved["dst"])) + rows = [moved] + sample.transfers[1:] + sp.load_transfers(client, rows, 500) + failed = sp.failures(sp.check_transfers(client, rows)) + assert failed == [f"no transfer id held by more than one edge: expected 0, found 1 ({moved['id']})"] + + +# --- failing the right way --------------------------------------------------- + +def test_a_refused_statement_is_not_retryable(engine): + with pytest.raises(sp.SamyamaRejected) as e: + sp.SamyamaHttp(engine.url).query("THIS IS NOT CYPHER") + assert e.value.status == 400 + + +def test_an_unreachable_server_is_retryable(): + from conftest import free_port + client = sp.SamyamaHttp(f"http://127.0.0.1:{free_port()}", timeout_seconds=2) + with pytest.raises(sp.SamyamaUnavailable) as e: + client.query("RETURN 1") + assert sp.is_retryable(e.value) + + +@pytest.mark.parametrize("error, retryable", [ + (sp.SamyamaUnavailable("HTTP 503: writes refused"), True), + (sp.SamyamaRejected(400, "syntax"), False), + (sp.SamyamaRejected(401, "no credential"), False), + (sp.SamyamaError("wrote 59 transfers for 60 rows"), False), + (sp.SourceDataError("a.csv:3: empty 'holder'"), False), +]) +def test_only_an_unavailable_server_is_worth_retrying(error, retryable): + assert sp.is_retryable(error) == retryable + + +@pytest.mark.parametrize("status, retryable", [ + (400, False), (401, False), (403, False), (404, False), + (408, True), (429, True), (500, True), (503, True), +]) +def test_an_http_status_is_classified_as_the_scheduler_needs(status, retryable): + # The engine does not answer every status on demand, so a stub does: what + # is under test is the client's reading of the status, not the server. + import http.server + import json as _json + import threading + + class Stub(http.server.BaseHTTPRequestHandler): + def do_POST(self): + self.rfile.read(int(self.headers["Content-Length"])) + body = _json.dumps({"error": f"stub {status}"}).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + server = http.server.HTTPServer(("127.0.0.1", 0), Stub) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + with pytest.raises(sp.SamyamaError) as e: + sp.SamyamaHttp(f"http://127.0.0.1:{server.server_port}").query("RETURN 1") + assert sp.is_retryable(e.value) == retryable + assert f"stub {status}" in str(e.value), "the server's own message is kept" + finally: + server.shutdown() + server.server_close() + + +def test_a_missing_or_wrong_token_is_refused_and_the_right_one_works(secured_engine, sample): + for token in (None, "0" * 64): + with pytest.raises(sp.SamyamaRejected) as e: + sp.SamyamaHttp(secured_engine.url, token=token).query("RETURN 1") + assert e.value.status == 401 + load_all(sp.SamyamaHttp(secured_engine.url, token=secured_engine.token), sample, 500) + assert_graph_matches(secured_engine, sample) diff --git a/tests/ci_gate_enumeration.rs b/tests/ci_gate_enumeration.rs index 0aad12d3..6acb256e 100644 --- a/tests/ci_gate_enumeration.rs +++ b/tests/ci_gate_enumeration.rs @@ -36,11 +36,18 @@ const AGGREGATOR: &str = "ci-ok"; /// that is in neither fails. So an entry cannot outlive the decision it records, /// and a new suite cannot quietly escape the gate -- escaping it means adding a /// line here, with a reason, in a diff someone reads. -const UNGATED: &[(&str, &str)] = &[( - "gpml", - "conformance has never gated a merge; requiring it is a policy decision, \ - separate from folding it into this workflow (#1865)", -)]; +const UNGATED: &[(&str, &str)] = &[ + ( + "gpml", + "conformance has never gated a merge; requiring it is a policy decision, \ + separate from folding it into this workflow (#1865)", + ), + ( + "orchestration", + "a new suite that installs Airflow and Dagster from PyPI; making it block \ + every merge is a policy decision, separate from adding it (#1715)", + ), +]; fn jobs(doc: &serde_yaml::Value) -> Vec { doc["jobs"] From 185151131a0bf33e05af2a4165e6795e5d8c326e Mon Sep 17 00:00:00 2001 From: kalyanpulikanti_01 Date: Thu, 22 Oct 2026 22:43:19 +0530 Subject: [PATCH 3/4] Resolve Dagster worker credentials at execution and retry checks --- docs/ORCHESTRATION-DAGSTER.md | 5 ++- examples/orchestration/samyama_dagster.py | 21 ++++++---- examples/orchestration/tests/test_dagster.py | 43 +++++++++++++++++++- 3 files changed, 58 insertions(+), 11 deletions(-) diff --git a/docs/ORCHESTRATION-DAGSTER.md b/docs/ORCHESTRATION-DAGSTER.md index 06111650..1c382e77 100644 --- a/docs/ORCHESTRATION-DAGSTER.md +++ b/docs/ORCHESTRATION-DAGSTER.md @@ -45,8 +45,9 @@ is no Samyama integration package to install. | `SAMYAMA_URL` | required -- the server's HTTP endpoint, e.g. `https://graph.internal:8080` | | `SAMYAMA_TOKEN` | the token from step 1; leave unset for a server without an auth file | -Both are read through `dg.EnvVar`, so the token is resolved when a run starts -and is not stored with the code location or shown in the UI. +The URL uses `dg.EnvVar`; the optional token is read when the resource creates +its client on the execution worker, so worker-only secrets are available. The +token value is not stored with the code location or shown in the UI. ## 4. Run it diff --git a/examples/orchestration/samyama_dagster.py b/examples/orchestration/samyama_dagster.py index a0f2baf5..fb729dea 100644 --- a/examples/orchestration/samyama_dagster.py +++ b/examples/orchestration/samyama_dagster.py @@ -11,8 +11,9 @@ - `SAMYAMA_URL` (required) names the server's HTTP endpoint. `SAMYAMA_TOKEN` is the bearer token, needed only when the server runs with an auth file. Both - are read through `dg.EnvVar`, so the secret is resolved when a run starts and - never stored with the code location. + are resolved in the execution environment: the URL through `dg.EnvVar`, + and the optional token when the resource creates its client. No token value + is stored with the code location. - The source files and batch size are the `source_files` resource. They default to the bundled sample; set them per run in the launchpad's resource config. @@ -43,10 +44,14 @@ class SamyamaResource(dg.ConfigurableResource): url: str token: Optional[str] = None + token_env_var: Optional[str] = None timeout_seconds: float = 30.0 def client(self) -> sp.SamyamaHttp: - return sp.SamyamaHttp(self.url, token=self.token or None, timeout_seconds=self.timeout_seconds) + token = self.token + if token is None and self.token_env_var is not None: + token = os.environ.get(self.token_env_var) + return sp.SamyamaHttp(self.url, token=token or None, timeout_seconds=self.timeout_seconds) class SourceFiles(dg.ConfigurableResource): @@ -106,13 +111,13 @@ def _result(checks: list[sp.Check]) -> dg.AssetCheckResult: ) -@dg.asset_check(asset=accounts, blocking=True) +@dg.asset_check(asset=accounts, blocking=True, retry_policy=RETRY) @fail_fast_on_permanent_errors def accounts_match_source(samyama: SamyamaResource, source_files: SourceFiles) -> dg.AssetCheckResult: return _result(sp.check_accounts(samyama.client(), sp.read_accounts(source_files.accounts_csv))) -@dg.asset_check(asset=transfers, blocking=True) +@dg.asset_check(asset=transfers, blocking=True, retry_policy=RETRY) @fail_fast_on_permanent_errors def transfers_match_source(samyama: SamyamaResource, source_files: SourceFiles) -> dg.AssetCheckResult: return _result(sp.check_transfers(samyama.client(), sp.read_transfers(source_files.transfers_csv))) @@ -128,9 +133,9 @@ def build_definitions() -> dg.Definitions: resources={ "samyama": SamyamaResource( url=dg.EnvVar("SAMYAMA_URL"), - # A server without an auth file needs no token, and an EnvVar - # that is unset fails the run -- so it is attached only when set. - token=dg.EnvVar("SAMYAMA_TOKEN") if os.environ.get("SAMYAMA_TOKEN") else None, + # Resolve optional credentials on workers, where secrets may be + # injected after the code location builds these definitions. + token_env_var="SAMYAMA_TOKEN", ), "source_files": SourceFiles(), }, diff --git a/examples/orchestration/tests/test_dagster.py b/examples/orchestration/tests/test_dagster.py index a1665322..dd1beba9 100644 --- a/examples/orchestration/tests/test_dagster.py +++ b/examples/orchestration/tests/test_dagster.py @@ -9,6 +9,7 @@ import csv import dagster as dg +import pytest import samyama_dagster as sd from conftest import Source, assert_graph_matches @@ -110,8 +111,48 @@ def test_the_definitions_read_the_server_from_the_environment(engine, sample, mo def test_the_definitions_read_the_token_from_the_environment(secured_engine, sample, monkeypatch): monkeypatch.setenv("SAMYAMA_URL", secured_engine.url) - monkeypatch.setenv("SAMYAMA_TOKEN", secured_engine.token) + # Code locations may load before workers receive their secret environment. + monkeypatch.delenv("SAMYAMA_TOKEN", raising=False) job = sd.build_definitions().resolve_implicit_global_asset_job_def() + monkeypatch.setenv("SAMYAMA_TOKEN", secured_engine.token) result = job.execute_in_process(raise_on_error=False) assert result.success assert_graph_matches(secured_engine, sample) + + +@pytest.mark.parametrize("check_name", ["check_accounts", "check_transfers"]) +def test_transient_check_failure_is_retried_then_succeeds(engine, sample, monkeypatch, check_name): + real = getattr(sd.sp, check_name) + attempts = [] + + def transient_once(*args, **kwargs): + attempts.append(1) + if len(attempts) == 1: + raise sd.sp.SamyamaUnavailable("HTTP 503: temporarily unavailable") + return real(*args, **kwargs) + + monkeypatch.setattr(sd.sp, check_name, transient_once) + # Exercise Dagster's real retry scheduler without waiting thirty seconds. + monkeypatch.setattr(dg.RetryPolicy, "calculate_delay", lambda self, attempt_num: 0) + result = run(resources(engine, sample)) + assert result.success + assert len(attempts) == 2 + retries = [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] + assert len(retries) == 1 + assert checks(result) == {"accounts_match_source": True, "transfers_match_source": True} + assert_graph_matches(engine, sample) + + +@pytest.mark.parametrize("check_name", ["check_accounts", "check_transfers"]) +def test_permanent_check_failure_is_not_retried(engine, sample, monkeypatch, check_name): + attempts = [] + + def rejected(*args, **kwargs): + attempts.append(1) + raise sd.sp.SamyamaRejected(401, "missing credential") + + monkeypatch.setattr(sd.sp, check_name, rejected) + result = run(resources(engine, sample)) + assert not result.success + assert len(attempts) == 1 + assert not [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] From b2274bfc9bedda1a9a64935c346693d1ed460d14 Mon Sep 17 00:00:00 2001 From: kalyanpulikanti_01 Date: Thu, 22 Oct 2026 22:46:29 +0530 Subject: [PATCH 4/4] Reject credential-bearing redirects and detect per-ID account duplicates --- examples/orchestration/samyama_pipeline.py | 29 +++++++-- examples/orchestration/tests/test_pipeline.py | 64 +++++++++++++++++++ 2 files changed, 86 insertions(+), 7 deletions(-) diff --git a/examples/orchestration/samyama_pipeline.py b/examples/orchestration/samyama_pipeline.py index 6c40cf00..8da59809 100644 --- a/examples/orchestration/samyama_pipeline.py +++ b/examples/orchestration/samyama_pipeline.py @@ -78,6 +78,13 @@ class SourceDataError(ValueError): """The input files are wrong; a retry will read the same files.""" +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """API redirects must not forward write credentials to a different server.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + @dataclass class SamyamaHttp: """A minimal client for `POST /api/query`. @@ -114,7 +121,8 @@ def query(self, cypher: str, params: Optional[dict] = None, read_only: bool = Fa method="POST", ) try: - with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response: + with urllib.request.build_opener(_NoRedirect()).open( + request, timeout=self.timeout_seconds) as response: return json.loads(response.read()) except urllib.error.HTTPError as e: message = _error_message(e) @@ -277,6 +285,14 @@ def batches(rows: Sequence[dict], size: int) -> Iterator[list[dict]]: """ +DUPLICATED_ACCOUNT_IDS = """ +MATCH (a:Account) +WITH a.id AS id, count(a) AS n +WHERE n > 1 +RETURN collect(id) AS duplicated +""" + + def ensure_schema(client: SamyamaHttp) -> int: for statement in SCHEMA_STATEMENTS: client.query(statement) @@ -326,15 +342,14 @@ def passed(self) -> bool: def check_accounts(client: SamyamaHttp, rows: Sequence[dict]) -> list[Check]: - ids = [r["id"] for r in rows] + ids = {r["id"] for r in rows} + duplicated = sorted(ids.intersection(client.scalar(DUPLICATED_ACCOUNT_IDS))) return [ Check("accounts present with the file's values", len(rows), client.scalar(CHECK_ACCOUNTS, {"rows": list(rows)})), - # A second node with the same id -- written by something that used - # CREATE rather than MERGE -- passes the check above and fails this one. - Check("no account id held by more than one node", len(set(ids)), - client.scalar("UNWIND $ids AS id MATCH (a:Account {id: id}) RETURN count(a)", - {"ids": ids})), + # Count duplicates by ID: an extra copy of A must not cancel a missing B. + Check("no account id held by more than one node", 0, len(duplicated), + ", ".join(duplicated[:20])), ] diff --git a/examples/orchestration/tests/test_pipeline.py b/examples/orchestration/tests/test_pipeline.py index 3ff38643..df94e142 100644 --- a/examples/orchestration/tests/test_pipeline.py +++ b/examples/orchestration/tests/test_pipeline.py @@ -266,3 +266,67 @@ def test_a_missing_or_wrong_token_is_refused_and_the_right_one_works(secured_eng assert e.value.status == 401 load_all(sp.SamyamaHttp(secured_engine.url, token=secured_engine.token), sample, 500) assert_graph_matches(secured_engine, sample) + + +def test_duplicate_and_missing_account_cannot_cancel_each_other(engine, sample): + a, b = sample.accounts[:2] + _create_account(engine, a) + _create_account(engine, a) + checked = sp.check_accounts(sp.SamyamaHttp(engine.url), [a, b]) + # The old total-count check returned two for both checks and passed. + assert checked[0].passed + failed = sp.failures(checked) + assert len(failed) == 1 and "more than one node" in failed[0] + assert a["id"] in failed[0] + + +@pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) +def test_redirect_cannot_forward_write_credentials(status): + import http.server + import json + import threading + + source_auth, redirected_auth = [], [] + + class Receiver(http.server.BaseHTTPRequestHandler): + def do_GET(self): + redirected_auth.append(self.headers.get("Authorization")) + body = json.dumps({"records": [[1]]}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_POST = do_GET + + def log_message(self, *args): + pass + + receiver = http.server.HTTPServer(("127.0.0.1", 0), Receiver) + + class Redirect(http.server.BaseHTTPRequestHandler): + def do_POST(self): + source_auth.append(self.headers.get("Authorization")) + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(status) + self.send_header("Location", f"http://127.0.0.1:{receiver.server_port}/capture") + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + source = http.server.HTTPServer(("127.0.0.1", 0), Redirect) + for server in (source, receiver): + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + client = sp.SamyamaHttp(f"http://127.0.0.1:{source.server_port}", token="test-only-token") + with pytest.raises(sp.SamyamaRejected) as error: + client.query("RETURN 1") + assert error.value.status == status + assert source_auth == ["Bearer test-only-token"] + assert redirected_auth == [] + finally: + for server in (source, receiver): + server.shutdown() + server.server_close()