Refactor node/edge iterator locking

.github/workflows/bench-graphql.yml

@@ -38,10 +38,53 @@ jobs:
           k6-version: "1.0.0"
       - name: Run GraphQL benchmarks
         run: cd graphql-bench && make bench-local
+      # Failure only, so green runs stay quiet. `make bench-local` runs the server and k6 under
+      # `concurrently --kill-others`, so a server that died (or was still building its bench graph
+      # when k6's setup() tried to connect) shows up only in these logs.
+      - name: Print bench diagnostics
+        if: failure()
+        run: |
+          cd graphql-bench || exit 0
+          echo "::group::output.json"
+          cat output.json || echo "(no output.json was produced)"
+          echo "::endgroup::"
+          echo "::group::raphtory server log (last 300 lines)"
+          tail -n 300 server.log || echo "(no server.log)"
+          echo "::endgroup::"
+          echo "::group::k6 log (last 300 lines)"
+          # gzipped as it is written; a long run's log is far too big to keep uncompressed
+          gzip -dc k6.log.gz 2>/dev/null | tail -n 300 || echo "(no k6.log.gz)"
+          echo "::endgroup::"
+          echo "::group::k6 csv output"
+          ls -la output.csv.gz || echo "(no output.csv.gz)"
+          echo "rows: $(gzip -dc output.csv.gz 2>/dev/null | wc -l || echo unknown)"
+          echo "--- header + first rows ---"
+          { gzip -dc output.csv.gz 2>/dev/null || true; } | head -n 5 || true
+          echo "::endgroup::"
+      - name: Upload bench diagnostics
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: graphql-bench-diagnostics
+          path: |
+            graphql-bench/output.json
+            graphql-bench/output.csv.gz
+            graphql-bench/build.log
+            graphql-bench/server.log
+            graphql-bench/k6.log.gz
+          if-no-files-found: warn
+          retention-days: 7
+      # github-action-benchmark below has to `git switch gh-pages`, which aborts if any tracked
+      # file is dirty. The bench dirties several: Cargo.lock from the build, pnpm-lock.yaml from
+      # `pnpm install --no-frozen-lockfile`, and graphql-bench/data/apache/master from the server
+      # writing to the graph it serves. Restoring them path by path meant each newly dirtied file
+      # broke the run, so reset the whole tree instead. This only touches tracked files, so the
+      # bench outputs the later steps read (output.json above all) are left alone.
       - name: Restore modified files
         run: |
-          git restore Cargo.lock # modified by build; github-action-benchmark can't switch to gh-pages with dirty working tree
-          git restore graphql-bench/data/apache/master # otherwise github-action-benchmark fails to create the commit
+          git reset --hard
+          echo "worktree after reset (empty means clean):"
+          git status --porcelain
       - name: Print bench results
         run: cat graphql-bench/output.json
       - name: Store benchmark results from master branch

db4-storage/src/pages/edge_store.rs

@@ -1,7 +1,7 @@
 use super::{edge_page::writer::EdgeWriter, resolve_pos};
 use crate::{
     LocalPOS,
-    api::edges::{EdgeRefOps, EdgeSegmentOps, LockedESegment},
+    api::edges::{EdgeEntryOps, EdgeRefOps, EdgeSegmentOps, LockedESegment},
     error::StorageError,
     pages::{
         SegmentCounts,
@@ -644,13 +644,24 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
         })
     }
 
-    pub fn iter(&self, layer: LayerId) -> impl Iterator<Item = ES::Entry<'_>> + '_ {
+    pub fn iter(
+        &self,
+        layer: LayerId,
+    ) -> impl Iterator<Item = SegmentLockedEdgeEntry<ES, EXT>> + '_ {
         (0..self.segments.count())
             .filter_map(move |page_id| self.segments.get(page_id))
             .flat_map(move |page| {
-                (0..page.num_edges()).filter_map(move |local_edge| {
-                    page.layer_entry(LocalPOS(local_edge), layer, Some(page.head()))
-                })
+                let locked = Arc::new(page.locked());
+                (0..locked.num_edges())
+                    .map(LocalPOS)
+                    .map(move |pos| SegmentLockedEdgeEntry {
+                        locked: locked.clone(),
+                        pos,
+                    })
+                    .filter(move |edge| {
+                        let edge_ref: <<ES as EdgeSegmentOps>::ArcLockedSegment as LockedESegment>::EntryRef<'_> = edge.as_ref();
+                        edge_ref.has_layer_inner(layer)
+                    })
             })
     }
 
@@ -685,3 +696,29 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
         self.par_iter_segments().try_for_each(|seg| seg.flush())
     }
 }
+
+#[derive(Debug)]
+pub struct SegmentLockedEdgeEntry<
+    ES: EdgeSegmentOps<Extension = EXT>,
+    EXT: PersistenceStrategy<ES = ES>,
+> {
+    pos: LocalPOS,
+    locked: Arc<ES::ArcLockedSegment>,
+}
+
+impl<'a, ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>> EdgeEntryOps<'a>
+    for SegmentLockedEdgeEntry<ES, EXT>
+{
+    type Ref<'b>
+        = <ES::ArcLockedSegment as LockedESegment>::EntryRef<'b>
+    where
+        'a: 'b,
+        Self: 'b;
+
+    fn as_ref<'b>(&'b self) -> Self::Ref<'b>
+    where
+        'a: 'b,
+    {
+        self.locked.entry_ref(self.pos, None)
+    }
+}

db4-storage/src/pages/node_store.rs

@@ -3,7 +3,7 @@ use crate::{
     LocalPOS,
     api::{
         node_type_index::NodeTypeIndexOps,
-        nodes::{LockedNSSegment, NodeSegmentOps},
+        nodes::{LockedNSSegment, NodeEntryOps, NodeSegmentOps},
     },
     error::StorageError,
     pages::{
@@ -181,6 +181,22 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>>
         })
     }
 
+    pub fn node_entries(&self) -> impl Iterator<Item = SegmentLockedNodeEntry<NS, EXT>> {
+        let count = self.segments.count();
+        (0..count).flat_map(|id| {
+            let ns = self
+                .get_segment(id)
+                .expect("segment should exist given count");
+            let locked = Arc::new(ns.locked());
+            (0..locked.num_nodes()).map(LocalPOS).map(
+                move |pos| SegmentLockedNodeEntry::<NS, EXT> {
+                    locked: locked.clone(),
+                    pos,
+                },
+            )
+        })
+    }
+
     pub fn num_segments(&self) -> usize {
         self.segments.count()
     }
@@ -189,10 +205,6 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>>
         self.segments.iter().map(|(_, page)| page.t_len()).sum()
     }
 
-    // pub fn segments(&self) -> &boxcar::Vec<Arc<NS>> {
-    //     &self.segments
-    // }
-
     pub fn segments_par_iter(&self) -> impl ParallelIterator<Item = &NS> {
         let len = self.segments.count();
         (0..len)
@@ -214,6 +226,31 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>>
     }
 }
 
+pub struct SegmentLockedNodeEntry<
+    NS: NodeSegmentOps<Extension = EXT>,
+    EXT: PersistenceStrategy<NS = NS>,
+> {
+    pos: LocalPOS,
+    locked: Arc<NS::ArcLockedSegment>,
+}
+
+impl<'a, NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>> NodeEntryOps<'a>
+    for SegmentLockedNodeEntry<NS, EXT>
+{
+    type Ref<'b>
+        = <NS::ArcLockedSegment as LockedNSSegment>::EntryRef<'b>
+    where
+        'a: 'b,
+        Self: 'b;
+
+    fn as_ref<'b>(&'b self) -> Self::Ref<'b>
+    where
+        'a: 'b,
+    {
+        self.locked.entry_ref(self.pos)
+    }
+}
+
 impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>>
     NodeStorageInner<NS, EXT>
 {
@@ -232,7 +269,7 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>>
             segments: boxcar::Vec::new(),
             stats: GraphStats::new().into(),
             node_type_index,
-            free_segments: free_segments.try_into().unwrap(),
+            free_segments,
             nodes_path,
             node_meta,
             edge_meta,

docs/reference/graphql/graphql_API.md

@@ -1255,7 +1255,7 @@ Target node ids.
 <td valign="top"><a href="#string">String</a></td>
 <td>
 
-Edge property to use as weight. If unset, all edges have weight 1.
+Edge property to use as weight.
 
 </td>
 </tr>
@@ -1266,6 +1266,15 @@ Edge property to use as weight. If unset, all edges have weight 1.
 
 Edge direction to follow. Defaults to BOTH.
 
+</td>
+</tr>
+<tr>
+<td colspan="2" align="right" valign="top">defaultWeight</td>
+<td valign="top"><a href="#value">Value</a></td>
+<td>
+
+Weight for edges that do not have a weight. Used if `weight` is not specified or the edge does not have a value for that property. Defaults to 1.
+
 </td>
 </tr>
 <tr>
@@ -6583,6 +6592,11 @@ later builds reuse; omit it to keep the saved one.
 <td></td>
 </tr>
 <tr>
+<td colspan="2" align="right" valign="top">indexGid</td>
+<td valign="top"><a href="#boolean">Boolean</a></td>
+<td></td>
+</tr>
+<tr>
 <td colspan="2" valign="top"><strong id="mutablegraph.indexedproperties">indexedProperties</strong></td>
 <td valign="top">[<a href="#string">String</a>!]</td>
 <td>

graphql-bench/.gitignore

@@ -16,3 +16,12 @@ dist
 output.csv.gz
 data
 
+samples.jsonl
+slow_req.log
+stress-test-report.html
+
+# bench-local run logs
+build.log
+server.log
+k6.log
+k6.log.gz

graphql-bench/Makefile

@@ -5,13 +5,44 @@ K6_IP=$(shell terraform output k6_ip | jq -r '.')
 RAPHTORY_IP=$(shell terraform output raphtory_ip | jq -r '.')
 
 
+# Logs from the last run. Nothing in here reaches the terminal on a run that worked: the server
+# and the bench are both noisy (tsc/tsup, the graph load, k6's own summary), and `--kill-others`
+# means a server that dies takes k6 down with it, so what you want is the whole lot on disk and
+# surfaced only when something breaks. process-k6-output.py prints them when it finds no results,
+# and BENCH_VERBOSE=1 prints its report on a good run too.
+#
+# The k6 log is gzipped as it is written: a long run emits a lot of progress lines, and any
+# request/response sampling that is switched on emits a whole JSON blob per sample, which reached
+# 200MB uncompressed. Read it with `gzip -dc k6.log.gz | less` (or `| tail -n 100`).
+BUILD_LOG ?= build.log
+SERVER_LOG ?= server.log
+K6_LOG ?= k6.log.gz
+
 build:
-	pnpm install --frozen-lockfile
-	pnpm build
+	@{ pnpm install --no-frozen-lockfile && pnpm build ; } > $(BUILD_LOG) 2>&1 \
+		|| { echo ">>> build failed:"; cat $(BUILD_LOG); exit 1; }
+
+# Purely informational, so nothing in here is allowed to fail.
+bench-env:
+	@echo ">>> bench environment"
+	@k6 version || echo "k6 not on PATH"
+	@python -V || true
+	@python -c 'import pandas; print("pandas", pandas.__version__)' || echo "pandas not importable"
+	@echo ">>> graphs available to the server:"
+	@ls -la data/apache 2>&1 || true
 
 bench-local: build
-	pnpm concurrently --raw --kill-others --names 'raphtory,bench' 'python server.py' 'sleep 10 && k6 run --out csv=output.csv.gz dist/bench.js' || :
-	python process-k6-output.py
+	@rm -f output.csv.gz output.json $(SERVER_LOG) $(K6_LOG)
+	@$(MAKE) --no-print-directory bench-env 2>&1 | gzip > $(K6_LOG) || true
+	@k6 version > /dev/null 2>&1 || echo ">>> WARNING: k6 is not on PATH"
+	@python -c 'import pandas' > /dev/null 2>&1 \
+		|| echo ">>> WARNING: pandas is not importable, results cannot be processed"
+	@echo ">>> running the bench; $(K6_LOG) and $(SERVER_LOG) have the details"
+	@pnpm concurrently --raw --kill-others --names 'raphtory,bench' \
+		'python -u server.py > $(SERVER_LOG) 2>&1' \
+		'{ sh wait-for-server.sh && k6 run --out csv=output.csv.gz dist/bench.js ; } 2>&1 | gzip >> $(K6_LOG)' \
+		; echo ">>> concurrently exited with status $$?" | gzip >> $(K6_LOG)
+	@BENCH_SERVER_LOG=$(SERVER_LOG) BENCH_K6_LOG=$(K6_LOG) python process-k6-output.py
 
 # call this like: make bench-e2-standard-16
 bench-%:
@@ -25,6 +56,28 @@ bench: deploy push-graphs build-raphtory
 stress-test: build
 	K6_WEB_DASHBOARD=true K6_WEB_DASHBOARD_EXPORT=stress-test-report.html k6 run dist/stress-test.js
 
+# Samples request/response pairs into $(SAMPLES) as JSONL (appends across runs).
+# SAMPLE_RATE is the chance of recording any given pair; failures and requests slower than
+# SLOW_MS are recorded whatever the rate, so `SAMPLE_RATE=0 make stress-test-sampled` keeps just
+# those two. The slow ones are pretty printed into $(SLOW_LOG) when the run ends.
+SAMPLES ?= samples.jsonl
+SAMPLE_RATE ?= 0.01
+SLOW_MS ?= 1000
+SLOW_LOG ?= slow_req.log
+stress-test-sampled: build
+	SAMPLE_RATE=$(SAMPLE_RATE) SLOW_MS=$(SLOW_MS) K6_WEB_DASHBOARD=true K6_WEB_DASHBOARD_EXPORT=stress-test-report.html \
+		k6 run --log-format=raw --console-output=$(SAMPLES) dist/stress-test.js; \
+		status=$$?; python slow_requests.py $(SAMPLES) --output $(SLOW_LOG); exit $$status
+
+# Appends the slow requests from $(SAMPLES) to $(SLOW_LOG), slowest first: the query behind the
+# max the summary reports, its variables, and whether the time went on the server (waiting) or
+# somewhere between k6 and it. Entries already in the log are skipped, so this is safe to repeat.
+# N limits how many are appended (0 = all), MIN_MS overrides the run's own SLOW_MS threshold.
+N ?= 0
+MIN_MS ?= 0
+slowest:
+	python slow_requests.py $(SAMPLES) --output $(SLOW_LOG) --top $(N) --min-ms $(MIN_MS)
+
 stress-test-multiple: build # 12 x 10m = 2h
 	for _ in $(seq 1 12); do \
 		K6_WEB_DASHBOARD=true K6_WEB_DASHBOARD_EXPORT=stress-test-report.html k6 run --duration=10m dist/stress-test.js