diff --git a/.github/workflows/bench-gate.yml b/.github/workflows/bench-gate.yml deleted file mode 100644 index 55e15154..00000000 --- a/.github/workflows/bench-gate.yml +++ /dev/null @@ -1,181 +0,0 @@ -name: Performance Gate - -on: - push: - branches: [main] - paths: - - 'src/**' - - 'Cargo.toml' - - 'benches/**' - pull_request: - branches: [main] - paths: - - 'src/**' - - 'Cargo.toml' - - 'benches/**' - -env: - CARGO_TERM_COLOR: always - MOON_NO_URING: "1" - # Regression threshold: fail if any critical bench regresses beyond this % - REGRESSION_THRESHOLD: "5" - -jobs: - bench-regression: - name: Criterion Regression Check - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - with: - fetch-depth: 0 - - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - # Restore baseline from main branch (if available) - - name: Restore baseline - id: baseline - uses: actions/cache/restore@v4 - with: - path: target/criterion - key: criterion-baseline-main - - - name: Run critical benchmarks - run: | - cargo bench --no-default-features --features runtime-tokio,jemalloc \ - --bench get_hotpath \ - --bench dispatch_baseline \ - --bench resp_parsing \ - --bench pubsub_hotpath \ - --bench distance_bench \ - --bench hnsw_bench \ - --bench fwht_bench \ - --bench entry_memory \ - --bench compact_key \ - --bench bptree_memory \ - -- --output-format bencher 2>&1 | tee bench_results.txt - - - name: Check for benchmark failures - run: | - if [ ! -s bench_results.txt ]; then - echo "ERROR: Benchmark output is empty — benchmarks may have failed to run." - exit 1 - fi - if grep -qi 'error\|panicked\|FAILED' bench_results.txt; then - echo "ERROR: Benchmark run contained errors:" - grep -i 'error\|panicked\|FAILED' bench_results.txt - exit 1 - fi - echo "Benchmarks completed successfully." - - - name: Check for regressions - if: steps.baseline.outputs.cache-hit == 'true' && github.event_name == 'pull_request' - run: | - echo "Checking for regressions against main baseline..." - echo "" - - # Criterion stores results in target/criterion//new/estimates.json - # Parse bencher-format output for ns/iter values and compare - FAILED=0 - CRITICAL_BENCHES="get_hotpath dispatch_baseline resp_parsing" - - for bench in $CRITICAL_BENCHES; do - # Extract current ns/iter from bencher output - CURRENT=$(grep "^test ${bench}" bench_results.txt | grep -oP '[\d,]+(?= ns/iter)' | tr -d ',') - if [ -z "$CURRENT" ]; then - # Try alternate format: "bench_name time: [low est high]" - CURRENT=$(grep "${bench}" bench_results.txt | grep -oP '[\d.]+(?= ns)' | head -1) - fi - - # Look for baseline estimate from Criterion's cached data - BASELINE_FILE="target/criterion/${bench}/base/estimates.json" - if [ -f "$BASELINE_FILE" ]; then - BASELINE=$(python3 -c " - import json - with open('${BASELINE_FILE}') as f: - d = json.load(f) - print(int(d.get('mean', d.get('median', {})).get('point_estimate', 0))) - " 2>/dev/null || echo "") - else - BASELINE="" - fi - - if [ -n "$CURRENT" ] && [ -n "$BASELINE" ] && [ "$BASELINE" -gt 0 ] 2>/dev/null; then - DELTA=$(( (CURRENT - BASELINE) * 100 / BASELINE )) - if [ "$DELTA" -gt "$REGRESSION_THRESHOLD" ]; then - echo "REGRESSION: ${bench} — ${DELTA}% slower (${BASELINE} → ${CURRENT} ns/iter, threshold: ${REGRESSION_THRESHOLD}%)" - FAILED=1 - else - echo "OK: ${bench} — ${DELTA}% change (${BASELINE} → ${CURRENT} ns/iter)" - fi - else - echo "SKIP: ${bench} — no baseline available for comparison" - fi - done - - echo "" - if [ "$FAILED" -eq 1 ]; then - echo "FAILED: Critical benchmark regression detected. Fix the regression or update the baseline." - exit 1 - else - echo "PASSED: No critical regressions found." - fi - - - name: No baseline available (first run) - if: steps.baseline.outputs.cache-hit != 'true' && github.event_name == 'pull_request' - run: | - echo "::warning::No performance baseline cached from main branch yet. Regression check skipped. Baseline will be saved on next main branch push." - echo "NOTE: No baseline cached from main branch yet." - echo "Benchmark results recorded but regression check skipped." - echo "Baseline will be saved on next main branch push." - - # Save baseline on main branch pushes - - name: Save baseline - if: github.ref == 'refs/heads/main' && github.event_name == 'push' - uses: actions/cache/save@v4 - with: - path: target/criterion - key: criterion-baseline-main - - - name: Archive benchmark results - if: always() - uses: actions/upload-artifact@v4 - with: - name: bench-results - path: bench_results.txt - retention-days: 90 - - memory-regression: - name: RSS Memory Gate - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Install redis-tools - run: sudo apt-get install -y redis-tools - - name: Build release - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Measure RSS after 100K keys - env: - MOON_NO_URING: "1" - KEY_COUNT: "100000" - run: | - ./target/release/moon --port 6399 --shards 1 & - MOON_PID=$! - sleep 2 - redis-benchmark -h 127.0.0.1 -p 6399 -t set -n "${KEY_COUNT}" -r "${KEY_COUNT}" -q - sleep 1 - RSS_KB=$(awk '/VmRSS/ {print $2}' /proc/${MOON_PID}/status) - RSS_MB=$((RSS_KB / 1024)) - echo "RSS after ${KEY_COUNT} keys: ${RSS_MB} MB (${RSS_KB} KB)" - BASELINE_MB=150 - if [ "${RSS_MB}" -gt "${BASELINE_MB}" ]; then - echo "FAILED: RSS ${RSS_MB} MB exceeds baseline ${BASELINE_MB} MB" - kill ${MOON_PID} 2>/dev/null || true - exit 1 - fi - echo "PASSED: RSS ${RSS_MB} MB within baseline ${BASELINE_MB} MB" - kill ${MOON_PID} 2>/dev/null || true diff --git a/.github/workflows/changelog-gate.yml b/.github/workflows/changelog-gate.yml deleted file mode 100644 index 22b51cd9..00000000 --- a/.github/workflows/changelog-gate.yml +++ /dev/null @@ -1,35 +0,0 @@ -name: CHANGELOG Gate - -on: - pull_request: - branches: [main] - -jobs: - changelog-check: - name: Require CHANGELOG update - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - with: - fetch-depth: 0 - - - name: Check for CHANGELOG changes - env: - PR_LABELS: ${{ join(github.event.pull_request.labels.*.name, ',') }} - BASE_SHA: ${{ github.event.pull_request.base.sha }} - HEAD_SHA: ${{ github.event.pull_request.head.sha }} - run: | - # Skip if 'skip-changelog' label is present - if echo "${PR_LABELS}" | grep -qi 'skip-changelog'; then - echo "skip-changelog label found -- skipping CHANGELOG check" - exit 0 - fi - - # Check if CHANGELOG.md was modified - if git diff --name-only "${BASE_SHA}...${HEAD_SHA}" | grep -q '^CHANGELOG.md$'; then - echo "CHANGELOG.md updated -- gate passed" - else - echo "ERROR: CHANGELOG.md was not updated in this PR." - echo "Either update CHANGELOG.md or add the 'skip-changelog' label." - exit 1 - fi diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0f20cf3a..53f35269 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,113 +6,82 @@ on: pull_request: branches: [main] +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + env: CARGO_TERM_COLOR: always + MOON_NO_URING: "1" jobs: - fmt: - name: Format + # ── Fast gate: no compilation needed (~10s) ───────────────────────── + lint: + name: Lint runs-on: ubuntu-latest steps: - uses: actions/checkout@v6 + with: + fetch-depth: 0 - uses: dtolnay/rust-toolchain@1.94.1 with: components: rustfmt - run: cargo fmt --check - - clippy: - name: Clippy - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.1 - with: - components: clippy - - uses: Swatinem/rust-cache@v2 - - name: Clippy (default features) - run: cargo clippy -- -D warnings - - name: Clippy (tokio runtime) - run: cargo clippy --no-default-features --features runtime-tokio,jemalloc -- -D warnings - - test: - name: Test - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.1 - - uses: Swatinem/rust-cache@v2 - - run: cargo test --no-default-features --features runtime-tokio,jemalloc - timeout-minutes: 15 - env: - MOON_NO_URING: "1" - - safety-audit: - name: Safety Audit - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - name: Audit unsafe blocks for SAFETY comments - run: bash scripts/audit-unsafe.sh - - name: Audit unwrap/expect ratchet - run: bash scripts/audit-unwrap.sh - - changelog: - name: CHANGELOG check - if: github.event_name == 'pull_request' - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - with: - fetch-depth: 0 - - name: Check CHANGELOG.md touched or skip-changelog label present + - run: bash scripts/audit-unsafe.sh + - run: bash scripts/audit-unwrap.sh + - name: CHANGELOG check + if: github.event_name == 'pull_request' env: PR_LABELS: ${{ join(github.event.pull_request.labels.*.name, ',') }} + BASE_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} run: | if echo "$PR_LABELS" | grep -q 'skip-changelog'; then - echo "skip-changelog label found — skipping check" + echo "skip-changelog label found — skipping" exit 0 fi - if git diff origin/main...HEAD --name-only | grep -q CHANGELOG.md; then + if git diff --name-only "${BASE_SHA}...${HEAD_SHA}" | grep -q CHANGELOG.md; then echo "CHANGELOG.md updated" else - echo "::error::CHANGELOG.md not updated under [Unreleased]. Add a changelog entry or apply the 'skip-changelog' label." + echo "::error::CHANGELOG.md not updated. Add an entry or apply 'skip-changelog' label." exit 1 fi - supply-chain: - name: Supply Chain Security + # ── Main gate: clippy + test in one job, shared compilation ───────── + # Clippy checks compile the crate; test reuses those artifacts via + # cargo's incremental cache in the same runner. Saves ~5 min vs + # separate jobs that each compile from scratch. + check: + name: Check runs-on: ubuntu-latest steps: - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@1.94.1 + with: + components: clippy - uses: Swatinem/rust-cache@v2 - - name: Install cargo-deny and cargo-audit - run: cargo install cargo-deny cargo-audit - - name: cargo deny check - run: cargo deny check - - name: cargo audit - run: cargo audit + with: + # Cache key covers all 3 feature combos so incremental builds hit. + shared-key: check-${{ hashFiles('Cargo.lock') }} - redis-compat: - name: Redis Compat - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.1 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Run redis_compat integration tests - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - MOON_TEST_PORT=6399 cargo test --release --no-default-features --features runtime-tokio,jemalloc --test redis_compat -- --ignored + # Clippy (3 feature configurations — shares compilation between runs) + - name: Clippy (default) + run: cargo clippy -- -D warnings + - name: Clippy (graph) + run: cargo clippy --features graph -- -D warnings + - name: Clippy (tokio) + run: cargo clippy --no-default-features --features runtime-tokio,jemalloc -- -D warnings + + # Test — reuses .rlib artifacts from clippy's check compilation above. + # Only the test harness + test binaries need building (incremental). + - name: Test + run: cargo test --no-default-features --features runtime-tokio,jemalloc + timeout-minutes: 15 + - name: Test (graph) + run: "cargo test --no-default-features --features runtime-tokio,jemalloc,graph --lib graph::" timeout-minutes: 5 - env: - MOON_NO_URING: "1" + # ── MSRV — minimal build, different cache key ─────────────────────── msrv: name: MSRV (1.94) runs-on: ubuntu-latest @@ -120,4 +89,6 @@ jobs: - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@1.94.1 - uses: Swatinem/rust-cache@v2 + with: + shared-key: msrv-${{ hashFiles('Cargo.lock') }} - run: cargo build --no-default-features --features runtime-tokio diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 63a1c21c..7d5d7145 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -8,6 +8,10 @@ on: schedule: - cron: '0 6 * * 1' +concurrency: + group: codeql-${{ github.ref }} + cancel-in-progress: true + jobs: analyze: name: Analyze (Rust) diff --git a/.github/workflows/compat.yml b/.github/workflows/compat.yml deleted file mode 100644 index 574acf48..00000000 --- a/.github/workflows/compat.yml +++ /dev/null @@ -1,351 +0,0 @@ -name: Client Compatibility - -on: - pull_request: - branches: [main] - schedule: - - cron: '0 4 * * 1' - -jobs: - redis-py: - name: redis-py - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon (tokio) - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Start Moon - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - env: - MOON_NO_URING: "1" - - uses: actions/setup-python@v5 - with: - python-version: '3.12' - - name: Install redis-py - run: pip install redis - - name: Run compatibility tests - run: | - python -c " - import redis - r = redis.Redis(host='127.0.0.1', port=6399) - # Basic operations - r.set('test_key', 'test_value') - assert r.get('test_key') == b'test_value' - r.delete('test_key') - assert r.get('test_key') is None - # Hash - r.hset('hash_key', mapping={'f1': 'v1', 'f2': 'v2'}) - assert r.hget('hash_key', 'f1') == b'v1' - assert r.hlen('hash_key') == 2 - # List - r.rpush('list_key', 'a', 'b', 'c') - assert r.llen('list_key') == 3 - assert r.lrange('list_key', 0, -1) == [b'a', b'b', b'c'] - # Set - r.sadd('set_key', 'a', 'b', 'c') - assert r.scard('set_key') == 3 - assert r.sismember('set_key', 'a') - # Sorted set - r.zadd('zset_key', {'a': 1.0, 'b': 2.0, 'c': 3.0}) - assert r.zcard('zset_key') == 3 - # Pipeline - pipe = r.pipeline() - pipe.set('p1', 'v1') - pipe.set('p2', 'v2') - pipe.get('p1') - pipe.get('p2') - results = pipe.execute() - assert results == [True, True, b'v1', b'v2'] - # INFO - info = r.info() - assert 'redis_version' in info - print('redis-py: ALL TESTS PASSED') - " - - go-redis: - name: go-redis - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon (tokio) - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Start Moon - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - env: - MOON_NO_URING: "1" - - uses: actions/setup-go@v5 - with: - go-version: '1.22' - - name: Run go-redis smoke test - run: | - GOTEST_DIR=$(mktemp -d "${RUNNER_TEMP:-/tmp}/go-compat-XXXXXX") - cat > "$GOTEST_DIR/main.go" << 'GOEOF' - package main - import ( - "context" - "fmt" - "github.com/redis/go-redis/v9" - ) - func main() { - ctx := context.Background() - rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6399"}) - defer rdb.Close() - rdb.Set(ctx, "go_key", "go_value", 0) - val, _ := rdb.Get(ctx, "go_key").Result() - if val != "go_value" { panic("GET failed") } - rdb.HSet(ctx, "go_hash", "f1", "v1") - hval, _ := rdb.HGet(ctx, "go_hash", "f1").Result() - if hval != "v1" { panic("HGET failed") } - fmt.Println("go-redis: ALL TESTS PASSED") - } - GOEOF - cd "$GOTEST_DIR" && go mod init compat && go mod tidy && go run main.go - - ioredis: - name: ioredis (Node.js) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon (tokio) - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Start Moon - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - env: - MOON_NO_URING: "1" - - uses: actions/setup-node@v4 - with: - node-version: '22' - - name: Install ioredis - run: npm install ioredis - - name: Run compatibility tests - run: | - node -e " - const Redis = require('ioredis'); - (async () => { - const r = new Redis({ host: '127.0.0.1', port: 6399, lazyConnect: true }); - await r.connect(); - // SET / GET - await r.set('node_key', 'node_value'); - const v = await r.get('node_key'); - if (v !== 'node_value') throw new Error('GET failed'); - // HSET / HGET - await r.hset('node_hash', 'f1', 'v1'); - const hv = await r.hget('node_hash', 'f1'); - if (hv !== 'v1') throw new Error('HGET failed'); - // Pipeline - const pipe = r.pipeline(); - pipe.set('np1', 'pv1'); - pipe.set('np2', 'pv2'); - pipe.get('np1'); - pipe.get('np2'); - const results = await pipe.exec(); - if (results[2][1] !== 'pv1') throw new Error('pipeline GET failed'); - if (results[3][1] !== 'pv2') throw new Error('pipeline GET failed'); - console.log('ioredis: ALL TESTS PASSED'); - await r.quit(); - })(); - " - - redis-rs: - name: redis-rs (Rust) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon (tokio) - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Start Moon - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - env: - MOON_NO_URING: "1" - - name: Run redis-rs smoke test - run: | - TMPDIR=$(mktemp -d) - cat > "$TMPDIR/Cargo.toml" << 'RSEOF' - [package] - name = "moon-compat-redis-rs" - version = "0.1.0" - edition = "2021" - [dependencies] - redis = "0.27" - RSEOF - mkdir -p "$TMPDIR/src" - cat > "$TMPDIR/src/main.rs" << 'RSEOF' - use redis::Commands; - fn main() { - let client = redis::Client::open("redis://127.0.0.1:6399/").unwrap(); - let mut con = client.get_connection().unwrap(); - // SET / GET - let _: () = con.set("rs_key", "rs_value").unwrap(); - let v: String = con.get("rs_key").unwrap(); - assert_eq!(v, "rs_value"); - // HSET / HGET - let _: () = con.hset("rs_hash", "f1", "v1").unwrap(); - let hv: String = con.hget("rs_hash", "f1").unwrap(); - assert_eq!(hv, "v1"); - // Pipeline - let (r1, r2): (String, String) = redis::pipe() - .cmd("SET").arg("rp1").arg("pv1").ignore() - .cmd("SET").arg("rp2").arg("pv2").ignore() - .cmd("GET").arg("rp1") - .cmd("GET").arg("rp2") - .query(&mut con).unwrap(); - assert_eq!(r1, "pv1"); - assert_eq!(r2, "pv2"); - println!("redis-rs: ALL TESTS PASSED"); - } - RSEOF - cd "$TMPDIR" && cargo run --release - - hiredis: - name: hiredis (C) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon (tokio) - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Start Moon - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - env: - MOON_NO_URING: "1" - - name: Install hiredis - run: sudo apt-get install -y libhiredis-dev - - name: Run hiredis smoke test - run: | - cat > /tmp/compat_test.c << 'CEOF' - #include - #include - #include - #include - int main() { - redisContext *c = redisConnect("127.0.0.1", 6399); - if (c == NULL || c->err) { fprintf(stderr, "connect failed\n"); return 1; } - redisReply *r; - /* SET / GET */ - r = redisCommand(c, "SET c_key c_value"); - freeReplyObject(r); - r = redisCommand(c, "GET c_key"); - if (strcmp(r->str, "c_value") != 0) { fprintf(stderr, "GET failed\n"); return 1; } - freeReplyObject(r); - /* HSET / HGET */ - r = redisCommand(c, "HSET c_hash f1 v1"); - freeReplyObject(r); - r = redisCommand(c, "HGET c_hash f1"); - if (strcmp(r->str, "v1") != 0) { fprintf(stderr, "HGET failed\n"); return 1; } - freeReplyObject(r); - /* Pipeline */ - redisAppendCommand(c, "SET cp1 pv1"); - redisAppendCommand(c, "SET cp2 pv2"); - redisAppendCommand(c, "GET cp1"); - redisAppendCommand(c, "GET cp2"); - redisGetReply(c, (void**)&r); freeReplyObject(r); - redisGetReply(c, (void**)&r); freeReplyObject(r); - redisGetReply(c, (void**)&r); - if (strcmp(r->str, "pv1") != 0) { fprintf(stderr, "pipeline GET1 failed\n"); return 1; } - freeReplyObject(r); - redisGetReply(c, (void**)&r); - if (strcmp(r->str, "pv2") != 0) { fprintf(stderr, "pipeline GET2 failed\n"); return 1; } - freeReplyObject(r); - printf("hiredis: ALL TESTS PASSED\n"); - redisFree(c); - return 0; - } - CEOF - gcc -o /tmp/compat_test /tmp/compat_test.c -lhiredis - /tmp/compat_test - - jedis: - name: jedis (Java) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@1.94.0 - - uses: Swatinem/rust-cache@v2 - - name: Build Moon (tokio) - run: cargo build --release --no-default-features --features runtime-tokio,jemalloc - env: - MOON_NO_URING: "1" - - name: Start Moon - run: | - ./target/release/moon --port 6399 --shards 1 & - sleep 2 - env: - MOON_NO_URING: "1" - - uses: actions/setup-java@v4 - with: - distribution: 'temurin' - java-version: '21' - - name: Run jedis smoke test - env: - JEDIS_VERSION: "5.2.0" - SLF4J_VERSION: "2.0.16" - POOL_VERSION: "2.12.0" - GSON_VERSION: "2.11.0" - run: | - mkdir -p /tmp/jedis-test - curl -sL "https://repo1.maven.org/maven2/redis/clients/jedis/${JEDIS_VERSION}/jedis-${JEDIS_VERSION}.jar" -o /tmp/jedis-test/jedis.jar - curl -sL "https://repo1.maven.org/maven2/org/slf4j/slf4j-api/${SLF4J_VERSION}/slf4j-api-${SLF4J_VERSION}.jar" -o /tmp/jedis-test/slf4j-api.jar - curl -sL "https://repo1.maven.org/maven2/org/slf4j/slf4j-simple/${SLF4J_VERSION}/slf4j-simple-${SLF4J_VERSION}.jar" -o /tmp/jedis-test/slf4j-simple.jar - curl -sL "https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/${POOL_VERSION}/commons-pool2-${POOL_VERSION}.jar" -o /tmp/jedis-test/commons-pool2.jar - curl -sL "https://repo1.maven.org/maven2/com/google/gson/gson/${GSON_VERSION}/gson-${GSON_VERSION}.jar" -o /tmp/jedis-test/gson.jar - cat > /tmp/jedis-test/CompatTest.java << 'JEOF' - import redis.clients.jedis.Jedis; - import redis.clients.jedis.Pipeline; - import java.util.List; - - public class CompatTest { - public static void main(String[] args) { - try (Jedis jedis = new Jedis("127.0.0.1", 6399)) { - // SET / GET - jedis.set("java_key", "java_value"); - String v = jedis.get("java_key"); - assert "java_value".equals(v) : "GET failed"; - // HSET / HGET - jedis.hset("java_hash", "f1", "v1"); - String hv = jedis.hget("java_hash", "f1"); - assert "v1".equals(hv) : "HGET failed"; - // Pipeline - Pipeline p = jedis.pipelined(); - p.set("jp1", "pv1"); - p.set("jp2", "pv2"); - p.get("jp1"); - p.get("jp2"); - List results = p.syncAndReturnAll(); - assert "pv1".equals(results.get(2)) : "pipeline GET1 failed"; - assert "pv2".equals(results.get(3)) : "pipeline GET2 failed"; - System.out.println("jedis: ALL TESTS PASSED"); - } - } - } - JEOF - cd /tmp/jedis-test && javac -cp "jedis.jar:commons-pool2.jar:slf4j-api.jar:gson.jar" CompatTest.java - cd /tmp/jedis-test && java -ea -cp ".:jedis.jar:commons-pool2.jar:slf4j-api.jar:slf4j-simple.jar:gson.jar" CompatTest diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index fa395a7c..dc8b43f4 100644 --- a/.github/workflows/fuzz.yml +++ b/.github/workflows/fuzz.yml @@ -3,15 +3,20 @@ name: Fuzz on: pull_request: branches: [main] + types: [labeled] schedule: - cron: '0 2 * * *' +concurrency: + group: fuzz-${{ github.ref }} + cancel-in-progress: true + env: CARGO_TERM_COLOR: always jobs: fuzz-pr: - if: github.event_name == 'pull_request' + if: github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'ci-fuzz') name: Fuzz PR runs-on: ubuntu-latest strategy: @@ -20,7 +25,12 @@ jobs: target: - resp_parse - resp_parse_differential + - inline_parse - wal_v3_record + - gossip_deser + - acl_rule + - rdb_load + - cypher_parse steps: - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@nightly @@ -38,8 +48,8 @@ jobs: name: fuzz-corpus path: fuzz/corpus/ continue-on-error: true - - name: Run fuzzer (5m, multi-process) - run: cargo +nightly fuzz run "$TARGET" -- -max_total_time=300 -max_len=65536 -fork=2 -ignore_crashes=0 + - name: Run fuzzer (15m, multi-process) + run: cargo +nightly fuzz run "$TARGET" -- -max_total_time=900 -max_len=65536 -fork=2 -ignore_crashes=0 env: MOON_NO_URING: "1" TARGET: ${{ matrix.target }} @@ -59,6 +69,7 @@ jobs: - gossip_deser - acl_rule - rdb_load + - cypher_parse steps: - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@nightly diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml new file mode 100644 index 00000000..72a0aacb --- /dev/null +++ b/.github/workflows/integration-tests.yml @@ -0,0 +1,48 @@ +name: Integration Tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + types: [labeled] + +concurrency: + group: integration-${{ github.ref }} + cancel-in-progress: true + +env: + CARGO_TERM_COLOR: always + MOON_NO_URING: "1" + +jobs: + durability: + name: Durability Tests + if: github.event_name == 'push' || contains(github.event.pull_request.labels.*.name, 'ci-full') + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@1.94.1 + - uses: Swatinem/rust-cache@v2 + - name: Build Moon (release) + run: cargo build --release --no-default-features --features runtime-tokio,jemalloc + - name: Run crash matrix tests + run: cargo test --release --no-default-features --features runtime-tokio,jemalloc --test durability_tests + timeout-minutes: 10 + - name: Run jepsen-lite tests + run: cargo test --release --no-default-features --features runtime-tokio,jemalloc --test jepsen_lite + timeout-minutes: 10 + + replication: + name: Replication Tests + if: github.event_name == 'push' || contains(github.event.pull_request.labels.*.name, 'ci-full') + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@1.94.1 + - uses: Swatinem/rust-cache@v2 + - name: Build Moon (release) + run: cargo build --release --no-default-features --features runtime-tokio,jemalloc + - name: Run replication hardening tests + run: cargo test --release --no-default-features --features runtime-tokio,jemalloc --test replication_hardening + timeout-minutes: 10 diff --git a/.gitignore b/.gitignore index 2208a564..280fb972 100644 --- a/.gitignore +++ b/.gitignore @@ -81,3 +81,4 @@ ssh libnull.rlib fuzz shard-*/ +.serena/ \ No newline at end of file diff --git a/.planning b/.planning index 598ce363..5aa77924 160000 --- a/.planning +++ b/.planning @@ -1 +1 @@ -Subproject commit 598ce363ab0493be999761ad0e7956209c54339e +Subproject commit 5aa77924d71f5c91136a6d10c792564652c80b8a diff --git a/CHANGELOG.md b/CHANGELOG.md index 5fbbdde8..b0041448 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added — Graph Engine Integration (v0.1.4, 2026-04-11) + +- **Property graph engine** (`src/graph/`, feature-gated under `graph`): segment-aligned CSR storage with SlotMap generational indices, ArcSwap lock-free reads, Roaring validity bitmaps, and Rabbit Order compaction for cache locality. 8,500+ LOC, 319 tests. +- **12 GRAPH.\* commands**: CREATE, ADDNODE, ADDEDGE, NEIGHBORS, QUERY, RO_QUERY, EXPLAIN, VSEARCH, HYBRID, INFO, LIST, DELETE — all with RESP3 Map responses and ACL annotations. +- **Cypher subset parser**: hand-rolled recursive descent with logos lexer, 12 clauses (MATCH/WHERE/RETURN/CREATE/DELETE/SET/MERGE/WITH/UNWIND/CALL/ORDER/LIMIT), parameterized queries ($param), nesting depth limit (64), plan caching. +- **Hybrid graph+vector queries**: graph-filtered vector search, vector-to-graph expansion, vector-guided walk with automatic strategy selection. +- **Traversal engine**: BFS/DFS/Dijkstra with bounded frontiers (100K cap), temporal decay + distance scoring, segment merge reader across mutable + immutable segments. +- **Graph indexes**: per-label/type Roaring bitmaps, boomphf minimal perfect hash (~3 bits/key), property B-tree for range queries. +- **Cross-shard traversal**: scatter-gather via SPSC mesh, graph hash tags for shard co-location, snapshot-LSN forwarding, configurable depth limit. +- **Graph MVCC**: extends existing TransactionManager with graph write intents, snapshot-isolated multi-hop traversal, bounded epoch hold (30s). +- **Graph WAL durability**: RESP-encoded graph commands in per-shard WAL, two-pass replay (nodes before edges), CRC32-validated CSR segment persistence. +- **Cost-based planner**: GraphStats with incremental degree tracking, graph-first vs vector-first strategy selection, P99 hub detection. +- **Criterion benchmarks**: CSR 1-hop 1.02ns, edge insert 64.8ns, 2-hop BFS 4.99µs, CSR freeze 5.12ms, SIMD cosine 384d 33.9ns. +- **Fair comparison benchmark** (`tests/graph_bench_compare.rs`): Moon 2.4x FalkorDB on Cypher MATCH, 19x on native 1-hop, 23x on population. +- **New dependencies**: `slotmap` 1.x (generational indices), `boomphf` 0.6 (MPH), `logos` 0.14 (Cypher lexer, optional). + +### Fixed — Deep Review Findings (2026-04-11) + +- **DoS protection**: `execute_profile` and `execute_mut` Cypher paths now enforce MAX_HOPS_LIMIT=20 and MAX_RESULT_ROWS=100K (were unbounded). +- **WAL correctness**: Cypher DELETE passes actual LSN to `remove_node`/`remove_edge` (was hardcoded to 0). +- **GRAPH.DROP metadata**: added missing phf dispatch table entry. +- **SAFETY comments**: added to all 7 unsafe SIMD/mmap functions. +- **BFS 30% faster**: scratch buffer reuse in SegmentMergeReader, zero-alloc CsrStorage callback, MergedNeighbor derives Copy. +- **ParallelBfs**: uses plain HashSet on sequential path (was DashSet with 64 shards overhead). +- **Recovery hardening**: CSR manifest path traversal validation, WAL embedding dimension cap (65536), LSN saturating_add. +- **CI optimized**: consolidated 26 jobs → 4 per PR, concurrency groups cancel superseded runs, fixed org runner group for public repos. + ### Fixed — Wave 0-4 Gap Closure (2026-04-09) - **ZREVRANGEBYSCORE/ZREVRANGEBYLEX correctness bug:** Fixed double-swap of min/max bounds in `zrange_by_score` and `zrange_by_lex` that caused empty results for finite score ranges (e.g., `ZREVRANGEBYSCORE key 3 1`). Added finite-range test to `test-commands.sh`. diff --git a/Cargo.lock b/Cargo.lock index 280787d3..9c2a2ece 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -181,6 +181,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "beef" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1" + [[package]] name = "bitflags" version = "1.3.2" @@ -211,6 +217,18 @@ dependencies = [ "objc2", ] +[[package]] +name = "boomphf" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "617e2d952880a00583ddb9237ac3965732e8df6a92a8e7bcc054100ec467ec3b" +dependencies = [ + "crossbeam-utils", + "log", + "rayon", + "wyhash", +] + [[package]] name = "bstr" version = "1.12.1" @@ -1271,6 +1289,39 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "logos" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7251356ef8cb7aec833ddf598c6cb24d17b689d20b993f9d11a3d764e34e6458" +dependencies = [ + "logos-derive", +] + +[[package]] +name = "logos-codegen" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59f80069600c0d66734f5ff52cc42f2dabd6b29d205f333d61fd7832e9e9963f" +dependencies = [ + "beef", + "fnv", + "lazy_static", + "proc-macro2", + "quote", + "regex-syntax", + "syn", +] + +[[package]] +name = "logos-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24fb722b06a9dc12adb0963ed585f19fc61dc5413e6a9be9422ef92c091e731d" +dependencies = [ + "logos-codegen", +] + [[package]] name = "loom" version = "0.7.2" @@ -1527,6 +1578,7 @@ dependencies = [ "atoi", "atomic-waker", "aws-lc-rs", + "boomphf", "bumpalo", "bytes", "clap", @@ -1548,6 +1600,7 @@ dependencies = [ "io-uring 0.7.11", "itoa", "libc", + "logos", "loom", "lz4_flex", "memchr", @@ -1573,6 +1626,7 @@ dependencies = [ "serde_json", "sha1_smol", "sha2", + "slotmap", "smallvec", "socket2 0.6.3", "tempfile", @@ -1963,6 +2017,12 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" + [[package]] name = "rand_core" version = "0.9.5" @@ -2329,6 +2389,15 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "slotmap" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdd58c3c93c3d278ca835519292445cb4b0d4dc59ccfdf7ceadaab3f8aeb4038" +dependencies = [ + "version_check", +] + [[package]] name = "smallvec" version = "1.15.1" @@ -3139,6 +3208,15 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "wyhash" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf6e163c25e3fac820b4b453185ea2dea3b6a3e0a721d4d23d75bd33734c295" +dependencies = [ + "rand_core 0.6.4", +] + [[package]] name = "xxhash-rust" version = "0.8.15" diff --git a/Cargo.toml b/Cargo.toml index f00eae97..a0448fb7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,9 @@ http-body-util = "0.1" tikv-jemallocator = { version = "0.6", optional = true } monoio = { version = "0.2", optional = true, features = ["sync", "bytes"] } cudarc = { version = "0.12", optional = true, default-features = false, features = ["cuda-version-from-build-system"] } +slotmap = "1" +boomphf = "0.6" +logos = { version = "0.14", optional = true } [features] # Platform-aware defaults: @@ -80,10 +83,14 @@ cudarc = { version = "0.12", optional = true, default-features = false, features # cargo build --no-default-features --features runtime-monoio,jemalloc # force Monoio default = ["runtime-monoio", "jemalloc"] jemalloc = ["dep:tikv-jemallocator"] -runtime-tokio = ["tokio/rt-multi-thread", "tokio/io-util", "tokio/signal", "tokio/time", "tokio/fs", "dep:tokio-util", "dep:tokio-rustls", "dep:aws-lc-rs", "dep:rustls", "rustls/aws_lc_rs", "dep:rustls-pemfile"] +runtime-tokio = ["tokio/rt-multi-thread", "tokio/io-util", "tokio/signal", "tokio/time", "tokio/fs", "tokio/process", "dep:tokio-util", "dep:tokio-rustls", "dep:aws-lc-rs", "dep:rustls", "rustls/aws_lc_rs", "dep:rustls-pemfile"] runtime-monoio = ["dep:monoio", "dep:monoio-rustls", "dep:aws-lc-rs", "dep:rustls", "rustls/aws_lc_rs", "dep:rustls-pemfile"] gpu-cuda = ["dep:cudarc"] simd-avx512 = [] +# OpenTelemetry exporter: reserves feature namespace for OTLP trace export. +# When wired, will gate tracing-opentelemetry + opentelemetry-otlp dependencies. +otel = [] +graph = ["dep:logos"] [target.'cfg(target_os = "linux")'.dependencies] io-uring = "0.7" @@ -146,3 +153,13 @@ harness = false [[bench]] name = "pubsub_hotpath" harness = false + +[[bench]] +name = "graph_bench" +harness = false +required-features = ["graph"] + +[[bench]] +name = "graph_traversal" +harness = false +required-features = ["graph"] diff --git a/benches/graph_bench.rs b/benches/graph_bench.rs new file mode 100644 index 00000000..7a3dd089 --- /dev/null +++ b/benches/graph_bench.rs @@ -0,0 +1,365 @@ +//! Criterion benchmarks for graph operations. +//! +//! Validates performance targets (PERF-01 through PERF-06): +//! - 1-hop CSR neighbor lookup (degree 50): < 1us +//! - 2-hop BFS expansion: < 100us at 1K nodes +//! - Edge insertion into MemGraph: < 10us +//! - CSR freeze (64K edges): < 5ms +//! - Command-level overhead for ADDNODE, ADDEDGE, NEIGHBORS + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use smallvec::smallvec; +use std::hint::black_box; +use std::sync::Arc; + +use moon::graph::csr::CsrSegment; +use moon::graph::csr::storage::CsrStorage; +use moon::graph::memgraph::MemGraph; +use moon::graph::simd; +use moon::graph::traversal::{BoundedBfs, SegmentMergeReader}; +use moon::graph::types::{Direction, NodeKey, PropertyMap}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Empty property map (no allocations). +fn empty_props() -> PropertyMap { + smallvec![] +} + +/// Build a MemGraph with `n` nodes, each connected to ~`degree` random neighbors. +/// Uses a deterministic LCG for reproducible benchmarks. +fn build_memgraph(n: usize, degree: usize) -> (MemGraph, Vec) { + let edge_threshold = n * degree + 1; // prevent auto-freeze + let mut g = MemGraph::new(edge_threshold); + + // Insert nodes. + let mut nodes = Vec::with_capacity(n); + for i in 0..n { + let nk = g.add_node(smallvec![0], empty_props(), None, i as u64 + 1); + nodes.push(nk); + } + + // Insert edges: deterministic pseudo-random via LCG. + let mut rng_state: u32 = 42; + for i in 0..n { + for _ in 0..degree { + rng_state = rng_state.wrapping_mul(1664525).wrapping_add(1013904223); + let target = (rng_state as usize) % n; + if target == i { + continue; // skip self-loops + } + let lsn = (i * degree + 1) as u64; + // Ignore errors (duplicate edges, etc.) -- we just want density. + let _ = g.add_edge(nodes[i], nodes[target], 1, 1.0, None, lsn); + } + } + + (g, nodes) +} + +/// Build a CSR segment from a MemGraph with `n` nodes and ~`degree` edges per node. +fn build_csr(n: usize, degree: usize) -> (CsrSegment, Vec) { + let (mut g, nodes) = build_memgraph(n, degree); + let frozen = g.freeze().expect("freeze should succeed"); + let csr = CsrSegment::from_frozen(frozen, 1_000_000).expect("CSR build should succeed"); + (csr, nodes) +} + +// --------------------------------------------------------------------------- +// PERF-01: 1-hop CSR neighbor lookup (degree 50) < 1us +// --------------------------------------------------------------------------- + +fn bench_csr_neighbor_1hop(c: &mut Criterion) { + let (csr, nodes) = build_csr(1000, 50); + // Pick a node in the middle for representative degree. + let target_key = nodes[500]; + let row = csr + .lookup_node(target_key) + .expect("node should exist in CSR"); + + c.bench_function("graph_neighbor_1hop_csr", |b| { + b.iter(|| { + let neighbors = black_box(&csr).neighbors_out(black_box(row)); + black_box(neighbors.len()) + }) + }); +} + +// --------------------------------------------------------------------------- +// PERF-01 (MemGraph variant): 1-hop neighbor lookup via MemGraph +// --------------------------------------------------------------------------- + +fn bench_memgraph_neighbor_1hop(c: &mut Criterion) { + let (g, nodes) = build_memgraph(1000, 50); + let target_key = nodes[500]; + + c.bench_function("graph_neighbor_1hop_memgraph", |b| { + b.iter(|| { + let count = black_box(&g) + .neighbors(black_box(target_key), Direction::Outgoing, u64::MAX) + .count(); + black_box(count) + }) + }); +} + +// --------------------------------------------------------------------------- +// PERF-02: 2-hop BFS expansion < 100us at 1K nodes +// --------------------------------------------------------------------------- + +fn bench_bfs_2hop(c: &mut Criterion) { + let mut group = c.benchmark_group("graph_expansion_2hop"); + + for &node_count in &[1_000usize, 10_000] { + let (g, nodes) = build_memgraph(node_count, 10); + let seed = nodes[0]; + let csr_segments: Vec> = Vec::new(); + + group.bench_with_input( + BenchmarkId::new("bfs_memgraph", node_count), + &node_count, + |b, _| { + b.iter(|| { + let reader = SegmentMergeReader::new( + Some(black_box(&g)), + &csr_segments, + Direction::Outgoing, + u64::MAX, + None, + ); + let bfs = BoundedBfs::new(2); + let result = bfs.execute(&reader, black_box(seed)); + black_box(result) + }) + }, + ); + } + + // Also test 2-hop on CSR segment. + for &node_count in &[1_000usize, 10_000] { + let (csr, nodes) = build_csr(node_count, 10); + let seed = nodes[0]; + let csr_segments = vec![Arc::new(CsrStorage::Heap(csr))]; + + group.bench_with_input( + BenchmarkId::new("bfs_csr", node_count), + &node_count, + |b, _| { + b.iter(|| { + let reader = SegmentMergeReader::new( + None, + &csr_segments, + Direction::Outgoing, + u64::MAX, + None, + ); + let bfs = BoundedBfs::new(2); + let result = bfs.execute(&reader, black_box(seed)); + black_box(result) + }) + }, + ); + } + + group.finish(); +} + +// --------------------------------------------------------------------------- +// PERF-04: Edge insertion into MemGraph < 10us +// --------------------------------------------------------------------------- + +fn bench_edge_insert(c: &mut Criterion) { + c.bench_function("graph_edge_insert", |b| { + b.iter_custom(|iters| { + let batch = iters.min(500_000) as usize; + let rounds = (iters as usize + batch - 1) / batch; + let mut total = std::time::Duration::ZERO; + for _ in 0..rounds { + let mut g = MemGraph::new(batch + 10); + let mut nodes = Vec::with_capacity(batch + 2); + for _ in 0..=(batch + 1) { + nodes.push(g.add_node(smallvec![0], empty_props(), None, 1)); + } + let start = std::time::Instant::now(); + for i in 0..batch { + let _ = black_box(g.add_edge( + nodes[i], + nodes[i + 1], + 1, + 1.0, + None, + black_box(i as u64 + 2), + )); + } + total += start.elapsed(); + } + total + }) + }); +} + +// --------------------------------------------------------------------------- +// PERF-05: CSR freeze (64K edges) < 5ms +// --------------------------------------------------------------------------- + +fn bench_csr_freeze(c: &mut Criterion) { + c.bench_function("graph_csr_freeze_64k", |b| { + b.iter_custom(|iters| { + let rounds = iters.min(200) as usize; + let mut total = std::time::Duration::ZERO; + for _ in 0..rounds { + // Build a MemGraph with ~64K edges. + let (mut g, _nodes) = build_memgraph(2000, 32); + let start = std::time::Instant::now(); + let frozen = g.freeze().expect("freeze ok"); + let csr = CsrSegment::from_frozen(frozen, 999); + black_box(&csr); + total += start.elapsed(); + } + // Scale to requested iters to keep Criterion happy. + total * (iters as u32) / (rounds as u32) + }) + }); +} + +// --------------------------------------------------------------------------- +// PERF-06: Command-level benchmarks (ADDNODE, ADDEDGE, NEIGHBORS) +// --------------------------------------------------------------------------- + +fn bench_addnode_command(c: &mut Criterion) { + c.bench_function("graph_addnode_command", |b| { + b.iter_custom(|iters| { + // Cap per-batch to avoid OOM on multi-billion iteration runs. + let batch = iters.min(500_000) as usize; + let rounds = (iters as usize + batch - 1) / batch; + let mut total = std::time::Duration::ZERO; + for _ in 0..rounds { + let mut g = MemGraph::new(batch + 1); + let start = std::time::Instant::now(); + for i in 0..batch { + let nk = g.add_node( + black_box(smallvec![1, 2]), + black_box(empty_props()), + None, + black_box(i as u64 + 1), + ); + black_box(nk); + } + total += start.elapsed(); + } + total + }) + }); +} + +fn bench_addedge_command(c: &mut Criterion) { + c.bench_function("graph_addedge_command", |b| { + b.iter_custom(|iters| { + let batch = iters.min(500_000) as usize; + let rounds = (iters as usize + batch - 1) / batch; + let mut total = std::time::Duration::ZERO; + for _ in 0..rounds { + let n = batch + 2; + let mut g = MemGraph::new(n + 1); + let mut nodes = Vec::with_capacity(n); + for i in 0..n { + nodes.push(g.add_node(smallvec![0], empty_props(), None, i as u64 + 1)); + } + let start = std::time::Instant::now(); + for i in 0..batch { + let _ = black_box(g.add_edge( + nodes[i], + nodes[i + 1], + 1, + 1.0, + None, + black_box(i as u64 + n as u64), + )); + } + total += start.elapsed(); + } + total + }) + }); +} + +fn bench_neighbors_command(c: &mut Criterion) { + let (g, nodes) = build_memgraph(1000, 50); + let target = nodes[500]; + + c.bench_function("graph_neighbors_command", |b| { + b.iter(|| { + let neighbors: Vec<_> = black_box(&g) + .neighbors(black_box(target), Direction::Both, u64::MAX) + .collect(); + black_box(neighbors.len()) + }) + }); +} + +// --------------------------------------------------------------------------- +// SIMD cosine similarity benchmarks +// --------------------------------------------------------------------------- + +fn bench_cosine_similarity(c: &mut Criterion) { + let mut group = c.benchmark_group("cosine_similarity"); + + // 384-dim: typical embedding size (MiniLM, etc.) + let dim = 384; + let a: Vec = (0..dim).map(|i| ((i as f32) * 0.01).sin()).collect(); + let b: Vec = (0..dim).map(|i| ((i as f32) * 0.02).cos()).collect(); + + group.bench_function("scalar_384d", |bench| { + bench.iter(|| { + black_box(simd::cosine_similarity_scalar_pub( + black_box(&a), + black_box(&b), + )) + }) + }); + + group.bench_function("simd_384d", |bench| { + bench.iter(|| black_box(simd::cosine_similarity(black_box(&a), black_box(&b)))) + }); + + // 768-dim: common for larger transformer models + let dim_768 = 768; + let a768: Vec = (0..dim_768).map(|i| ((i as f32) * 0.01).sin()).collect(); + let b768: Vec = (0..dim_768).map(|i| ((i as f32) * 0.02).cos()).collect(); + + group.bench_function("scalar_768d", |bench| { + bench.iter(|| { + black_box(simd::cosine_similarity_scalar_pub( + black_box(&a768), + black_box(&b768), + )) + }) + }); + + group.bench_function("simd_768d", |bench| { + bench.iter(|| black_box(simd::cosine_similarity(black_box(&a768), black_box(&b768)))) + }); + + group.finish(); +} + +// --------------------------------------------------------------------------- +// Criterion groups and main +// --------------------------------------------------------------------------- + +criterion_group!( + graph_benchmarks, + bench_csr_neighbor_1hop, + bench_memgraph_neighbor_1hop, + bench_bfs_2hop, + bench_edge_insert, + bench_csr_freeze, + bench_addnode_command, + bench_addedge_command, + bench_neighbors_command, + bench_cosine_similarity, +); + +criterion_main!(graph_benchmarks); diff --git a/benches/graph_traversal.rs b/benches/graph_traversal.rs new file mode 100644 index 00000000..e390e3c5 --- /dev/null +++ b/benches/graph_traversal.rs @@ -0,0 +1,143 @@ +//! Criterion benchmarks comparing parallel vs sequential BFS. +//! +//! Validates that `ParallelBfs` outperforms `BoundedBfs` on large-frontier +//! graphs where frontier exceeds PARALLEL_THRESHOLD (256 nodes). + +use criterion::{Criterion, criterion_group, criterion_main}; +use smallvec::smallvec; +use std::hint::black_box; +use std::sync::Arc; + +use moon::graph::csr::CsrStorage; +use moon::graph::memgraph::MemGraph; +use moon::graph::traversal::{BoundedBfs, ParallelBfs, SegmentMergeReader}; +use moon::graph::types::{Direction, NodeKey, PropertyMap}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Empty property map (no allocations). +fn empty_props() -> PropertyMap { + smallvec![] +} + +/// Build a MemGraph with `n` nodes, each connected to ~`degree` random +/// neighbors. Uses a deterministic LCG for reproducible benchmarks. +/// Identical to graph_bench.rs builder. +fn build_memgraph(n: usize, degree: usize) -> (MemGraph, Vec) { + let edge_threshold = n * degree + 1; // prevent auto-freeze + let mut g = MemGraph::new(edge_threshold); + + let mut nodes = Vec::with_capacity(n); + for i in 0..n { + let nk = g.add_node(smallvec![0], empty_props(), None, i as u64 + 1); + nodes.push(nk); + } + + // Deterministic pseudo-random via LCG. + let mut rng_state: u32 = 42; + for i in 0..n { + for _ in 0..degree { + rng_state = rng_state.wrapping_mul(1664525).wrapping_add(1013904223); + let target = (rng_state as usize) % n; + if target == i { + continue; + } + let lsn = (i * degree + 1) as u64; + let _ = g.add_edge(nodes[i], nodes[target], 1, 1.0, None, lsn); + } + } + + (g, nodes) +} + +// --------------------------------------------------------------------------- +// Parallel vs Sequential BFS benchmark +// --------------------------------------------------------------------------- + +fn bench_parallel_vs_sequential_bfs(c: &mut Criterion) { + let mut group = c.benchmark_group("parallel_bfs"); + + // 10K nodes, degree 50 -> ~500K edges, frontier at depth 1 has ~50 nodes + // from node 0, depth 2 grows beyond 256 triggering parallel path. + let (g, nodes) = build_memgraph(10_000, 50); + let seed = nodes[500]; // middle node for representative connectivity + let csr_segments: Vec> = Vec::new(); + + // Pre-check: verify frontier is large enough for meaningful comparison. + { + let reader = SegmentMergeReader::new( + Some(&g), + &csr_segments, + Direction::Outgoing, + u64::MAX - 1, + None, + ); + let seq = BoundedBfs::new(3).execute(&reader, seed).expect("ok"); + eprintln!( + "BFS precheck: 10K nodes, degree 50, depth 3 -> {} nodes visited", + seq.visited.len() + ); + } + + group.bench_function("sequential_bfs_10k_depth3", |b| { + b.iter(|| { + let reader = SegmentMergeReader::new( + Some(&g), + &csr_segments, + Direction::Outgoing, + u64::MAX - 1, + None, + ); + let bfs = BoundedBfs::new(3); + let result = bfs.execute(&reader, black_box(seed)); + black_box(&result); + result + }) + }); + + group.bench_function("parallel_bfs_10k_depth3", |b| { + b.iter(|| { + let reader = SegmentMergeReader::new( + Some(&g), + &csr_segments, + Direction::Outgoing, + u64::MAX - 1, + None, + ); + let bfs = ParallelBfs::new(3); + let result = bfs.execute(&reader, black_box(seed)); + black_box(&result); + result + }) + }); + + // Verify correctness: both produce identical result sets. + { + let reader = SegmentMergeReader::new( + Some(&g), + &csr_segments, + Direction::Outgoing, + u64::MAX - 1, + None, + ); + let seq = BoundedBfs::new(3).execute(&reader, seed).expect("ok"); + let par = ParallelBfs::new(3).execute(&reader, seed).expect("ok"); + + use slotmap::Key; + let mut seq_keys: Vec = seq.visited.iter().map(|v| v.0.data().as_ffi()).collect(); + let mut par_keys: Vec = par.visited.iter().map(|v| v.0.data().as_ffi()).collect(); + seq_keys.sort(); + par_keys.sort(); + assert_eq!( + seq_keys, par_keys, + "parallel and sequential BFS must produce identical results" + ); + } + + group.finish(); +} + +criterion_group!(graph_traversal, bench_parallel_vs_sequential_bfs); +criterion_main!(graph_traversal); diff --git a/docs/log-schema.md b/docs/log-schema.md new file mode 100644 index 00000000..a23ba1c4 --- /dev/null +++ b/docs/log-schema.md @@ -0,0 +1,54 @@ +# Moon Log Schema + +Structured tracing fields emitted by Moon's `tracing` instrumentation. All fields use `tracing::instrument` attributes with explicit field names — no unbounded cardinality. + +## Sampling + +- **Default:** 1/1000 for normal spans, full capture on error +- **Override:** `RUST_LOG=moon=debug` for full tracing (development only) +- **Future:** `--trace-sample-rate` CLI flag (gated behind `otel` feature) + +## Span Fields + +### Connection Lifecycle + +| Span | Fields | Cardinality | +|------|--------|-------------| +| `handle_connection` | `peer_addr`, `client_id`, `shard_id` | bounded (IP + u64 + usize) | +| `handle_connection_sharded_monoio` | `peer_addr`, `client_id` | bounded | + +### Replication + +| Span | Fields | Cardinality | +|------|--------|-------------| +| `replication_handshake` | `replica_id`, `master_host` | bounded (u64 + hostname) | + +### Persistence + +| Span | Fields | Cardinality | +|------|--------|-------------| +| `aof_rewrite` | `seq` (manifest sequence) | bounded (u64) | +| `rotate_segment` | (no custom fields — uses function-level span) | N/A | + +### Vector Search + +| Span | Fields | Cardinality | +|------|--------|-------------| +| `compact_segment` | (no custom fields) | N/A | + +## Key Logging Rules + +1. **Keys are never logged verbatim.** If a key appears in a log line, it must be hashed (e.g., `xxh64(key)`) to prevent unbounded cardinality and PII exposure. +2. **Command names** are logged via `sanitize_cmd_label()` which maps to a fixed set of ~120 known commands + `"unknown"` catch-all. +3. **Error messages** from client commands are logged at `WARN` level with the command name but not the key or arguments. +4. **Shard IDs** are small integers (0..num_shards), bounded by server config. + +## Log Levels + +| Level | Usage | +|-------|-------| +| `ERROR` | Unrecoverable I/O failures, persistence corruption, TLS errors | +| `WARN` | Recoverable errors (malformed input, slow subscriber drops, AOF write failures) | +| `INFO` | Server lifecycle (startup, shutdown, config changes, WAL rotation) | +| `DEBUG` | Per-connection lifecycle, SPSC drain, replication state changes | +| `TRACE` | Per-frame parsing, per-command dispatch (extremely verbose) | diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml index 99550352..b7137c26 100644 --- a/fuzz/Cargo.toml +++ b/fuzz/Cargo.toml @@ -14,7 +14,7 @@ bytes = "1" [dependencies.moon] path = ".." default-features = false -features = ["runtime-tokio", "jemalloc"] +features = ["runtime-tokio", "jemalloc", "graph"] [dependencies.tempfile] version = "3" @@ -56,5 +56,10 @@ name = "acl_rule" path = "fuzz_targets/acl_rule.rs" doc = false +[[bin]] +name = "cypher_parse" +path = "fuzz_targets/cypher_parse.rs" +doc = false + [workspace] members = ["."] diff --git a/fuzz/fuzz_targets/cypher_parse.rs b/fuzz/fuzz_targets/cypher_parse.rs new file mode 100644 index 00000000..c0d2813b --- /dev/null +++ b/fuzz/fuzz_targets/cypher_parse.rs @@ -0,0 +1,17 @@ +#![no_main] +//! Fuzz target for the Cypher parser. +//! +//! Feeds random bytes to `parse_cypher()` — must never panic. +//! Errors are expected (most random input is invalid Cypher); the goal +//! is to verify no panics, no stack overflows, no OOB reads. + +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|data: &[u8]| { + // The parser must handle any input gracefully — errors are fine, panics are not. + let _ = moon::graph::cypher::parse_cypher(data); + + // Also test with a small nesting depth limit to exercise the depth guard. + let mut parser = moon::graph::cypher::Parser::new(data, 4); + let _ = parser.parse(); +}); diff --git a/scripts/bench-graph-compare.sh b/scripts/bench-graph-compare.sh new file mode 100755 index 00000000..d3c9548b --- /dev/null +++ b/scripts/bench-graph-compare.sh @@ -0,0 +1,309 @@ +#!/usr/bin/env bash +set -euo pipefail + +############################################################################### +# bench-graph-compare.sh -- Moon Graph vs FalkorDB (Redis Graph) benchmark +# +# Compares graph operation throughput between Moon and FalkorDB using +# identical workloads over redis-cli. FalkorDB runs via Docker. +# +# Usage: +# ./scripts/bench-graph-compare.sh # Full run +# ./scripts/bench-graph-compare.sh --nodes 5000 # Custom scale +# ./scripts/bench-graph-compare.sh --skip-build # Skip Moon cargo build +# ./scripts/bench-graph-compare.sh --moon-only # Skip FalkorDB (no Docker) +############################################################################### + +PORT_MOON=16700 +PORT_FALKOR=16701 +NODES=1000 +EDGES=3000 +SKIP_BUILD=false +MOON_ONLY=false +BINARY="./target/release/moon" +MOON_PID="" +FALKOR_CID="" + +while [[ $# -gt 0 ]]; do + case "$1" in + --nodes) NODES="$2"; shift 2 ;; + --edges) EDGES="$2"; shift 2 ;; + --skip-build) SKIP_BUILD=true; shift ;; + --moon-only) MOON_ONLY=true; shift ;; + *) echo "Unknown: $1"; exit 1 ;; + esac +done + +log() { echo "[$(date '+%H:%M:%S')] $*"; } + +cleanup() { + set +e + [[ -n "${MOON_PID:-}" ]] && kill "$MOON_PID" 2>/dev/null && wait "$MOON_PID" 2>/dev/null + [[ -n "${FALKOR_CID:-}" ]] && docker stop "$FALKOR_CID" >/dev/null 2>&1 && docker rm "$FALKOR_CID" >/dev/null 2>&1 + set -e +} +trap cleanup EXIT + +CLI_MOON="redis-cli -p $PORT_MOON" +CLI_FALKOR="redis-cli -p $PORT_FALKOR" + +############################################################################### +# Build Moon +############################################################################### +if [[ "$SKIP_BUILD" == "false" ]]; then + log "Building Moon with graph feature..." + cargo build --release --no-default-features --features runtime-tokio,jemalloc,graph 2>&1 | tail -1 +fi + +############################################################################### +# Start Moon +############################################################################### +log "Starting Moon on port $PORT_MOON..." +MOON_NO_URING=1 $BINARY --port $PORT_MOON --shards 1 --protected-mode no > /tmp/moon_bench.log 2>&1 & +MOON_PID=$! +for i in $(seq 1 40); do + $CLI_MOON PING > /dev/null 2>&1 && break + sleep 0.25 +done +if ! $CLI_MOON PING > /dev/null 2>&1; then + log "ERROR: Moon failed to start" + cat /tmp/moon_bench.log + exit 1 +fi +log "Moon ready (PID=$MOON_PID)" + +############################################################################### +# Start FalkorDB (via Docker) +############################################################################### +if [[ "$MOON_ONLY" == "false" ]]; then + log "Starting FalkorDB on port $PORT_FALKOR..." + FALKOR_CID=$(docker run -d --rm -p $PORT_FALKOR:6379 falkordb/falkordb:latest 2>/dev/null || echo "") + if [[ -z "$FALKOR_CID" ]]; then + log "WARNING: Docker/FalkorDB not available. Running Moon-only benchmark." + MOON_ONLY=true + else + for i in $(seq 1 40); do + $CLI_FALKOR PING > /dev/null 2>&1 && break + sleep 0.25 + done + if ! $CLI_FALKOR PING > /dev/null 2>&1; then + log "WARNING: FalkorDB failed to start. Running Moon-only." + MOON_ONLY=true + else + log "FalkorDB ready (container=$FALKOR_CID)" + fi + fi +fi + +############################################################################### +# Benchmark function +############################################################################### +bench_engine() { + local ENGINE="$1" + local CLI="$2" + local PORT="$3" + local USE_CYPHER="$4" # "cypher" for FalkorDB, "native" for Moon + + log "=== [$ENGINE] Node Insertion ($NODES nodes) ===" + local START_NS=$(date +%s%N) + + if [[ "$USE_CYPHER" == "cypher" ]]; then + # FalkorDB uses GRAPH.QUERY with Cypher CREATE + for i in $(seq 1 $NODES); do + $CLI GRAPH.QUERY bench "CREATE (:Node {id: $i, name: 'knowledge_$i', confidence: 0.42})" > /dev/null 2>&1 + done + else + # Moon uses native GRAPH.ADDNODE + local NODE_IDS=() + for i in $(seq 1 $NODES); do + local LABEL="Concept" + case $((i % 5)) in + 1) LABEL="Fact" ;; + 2) LABEL="Event" ;; + 3) LABEL="Source" ;; + 4) LABEL="Agent" ;; + esac + local RESULT + RESULT=$($CLI GRAPH.ADDNODE bench "$LABEL" name "knowledge_$i" confidence "0.42" 2>&1) + NODE_IDS+=("$RESULT") + done + fi + + local END_NS=$(date +%s%N) + local ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) + local INSERT_OPS=$(( NODES * 1000 / (ELAPSED_MS + 1) )) + log " [$ENGINE] $NODES nodes in ${ELAPSED_MS}ms (${INSERT_OPS} inserts/s)" + eval "${ENGINE}_NODE_MS=$ELAPSED_MS" + eval "${ENGINE}_NODE_OPS=$INSERT_OPS" + + log "=== [$ENGINE] Edge Insertion ($EDGES edges) ===" + START_NS=$(date +%s%N) + local EDGE_OK=0 + + if [[ "$USE_CYPHER" == "cypher" ]]; then + # FalkorDB: MATCH two nodes, CREATE edge + for i in $(seq 1 $EDGES); do + local SRC=$(( (i % NODES) + 1 )) + local DST=$(( ((i * 7 + 3) % NODES) + 1 )) + [[ $SRC -eq $DST ]] && DST=$(( (DST % NODES) + 1 )) + local ETYPE="RELATED_TO" + case $((i % 5)) in + 1) ETYPE="DERIVED_FROM" ;; + 2) ETYPE="OBSERVED_AT" ;; + 3) ETYPE="SUPERSEDES" ;; + 4) ETYPE="CITED_BY" ;; + esac + $CLI GRAPH.QUERY bench "MATCH (a:Node {id: $SRC}), (b:Node {id: $DST}) CREATE (a)-[:${ETYPE} {weight: 0.42}]->(b)" > /dev/null 2>&1 && EDGE_OK=$((EDGE_OK + 1)) + done + else + # Moon: native GRAPH.ADDEDGE + for i in $(seq 1 $EDGES); do + local SRC_IDX=$(( i % ${#NODE_IDS[@]} )) + local DST_IDX=$(( (i * 7 + 3) % ${#NODE_IDS[@]} )) + [[ $SRC_IDX -eq $DST_IDX ]] && DST_IDX=$(( (DST_IDX + 1) % ${#NODE_IDS[@]} )) + local ETYPE="RELATED_TO" + case $((i % 5)) in + 1) ETYPE="DERIVED_FROM" ;; + 2) ETYPE="OBSERVED_AT" ;; + 3) ETYPE="SUPERSEDES" ;; + 4) ETYPE="CITED_BY" ;; + esac + $CLI GRAPH.ADDEDGE bench "${NODE_IDS[$SRC_IDX]}" "${NODE_IDS[$DST_IDX]}" "$ETYPE" WEIGHT "0.42" > /dev/null 2>&1 && EDGE_OK=$((EDGE_OK + 1)) + done + fi + + END_NS=$(date +%s%N) + ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) + local EDGE_OPS=$(( EDGES * 1000 / (ELAPSED_MS + 1) )) + log " [$ENGINE] $EDGES edges ($EDGE_OK ok) in ${ELAPSED_MS}ms (${EDGE_OPS} edges/s)" + eval "${ENGINE}_EDGE_MS=$ELAPSED_MS" + eval "${ENGINE}_EDGE_OPS=$EDGE_OPS" + + log "=== [$ENGINE] 1-Hop Queries (200 random) ===" + START_NS=$(date +%s%N) + local QUERY_OK=0 + + if [[ "$USE_CYPHER" == "cypher" ]]; then + for i in $(seq 1 200); do + local NID=$(( (RANDOM % NODES) + 1 )) + $CLI GRAPH.QUERY bench "MATCH (a:Node {id: $NID})-[r]->(b) RETURN b.id, type(r) LIMIT 50" > /dev/null 2>&1 && QUERY_OK=$((QUERY_OK + 1)) + done + else + for i in $(seq 1 200); do + local IDX=$(( RANDOM % ${#NODE_IDS[@]} )) + $CLI GRAPH.NEIGHBORS bench "${NODE_IDS[$IDX]}" > /dev/null 2>&1 && QUERY_OK=$((QUERY_OK + 1)) + done + fi + + END_NS=$(date +%s%N) + ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) + local QUERY_OPS=$(( 200 * 1000 / (ELAPSED_MS + 1) )) + local AVG_US=$(( ELAPSED_MS * 1000 / 200 )) + log " [$ENGINE] 200 queries in ${ELAPSED_MS}ms (${QUERY_OPS} qps, avg ${AVG_US}µs, $QUERY_OK ok)" + eval "${ENGINE}_QUERY_MS=$ELAPSED_MS" + eval "${ENGINE}_QUERY_OPS=$QUERY_OPS" + + log "=== [$ENGINE] 2-Hop Queries (100 random) ===" + START_NS=$(date +%s%N) + QUERY_OK=0 + + if [[ "$USE_CYPHER" == "cypher" ]]; then + for i in $(seq 1 100); do + local NID=$(( (RANDOM % NODES) + 1 )) + $CLI GRAPH.QUERY bench "MATCH (a:Node {id: $NID})-[*1..2]->(b) RETURN DISTINCT b.id LIMIT 100" > /dev/null 2>&1 && QUERY_OK=$((QUERY_OK + 1)) + done + else + for i in $(seq 1 100); do + local IDX=$(( RANDOM % ${#NODE_IDS[@]} )) + $CLI GRAPH.NEIGHBORS bench "${NODE_IDS[$IDX]}" DEPTH 2 > /dev/null 2>&1 && QUERY_OK=$((QUERY_OK + 1)) + done + fi + + END_NS=$(date +%s%N) + ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) + QUERY_OPS=$(( 100 * 1000 / (ELAPSED_MS + 1) )) + AVG_US=$(( ELAPSED_MS * 1000 / 100 )) + log " [$ENGINE] 100 queries in ${ELAPSED_MS}ms (${QUERY_OPS} qps, avg ${AVG_US}µs, $QUERY_OK ok)" + eval "${ENGINE}_2HOP_MS=$ELAPSED_MS" + eval "${ENGINE}_2HOP_OPS=$QUERY_OPS" + + log "=== [$ENGINE] Cypher Pattern Match (50 queries) ===" + START_NS=$(date +%s%N) + QUERY_OK=0 + + if [[ "$USE_CYPHER" == "cypher" ]]; then + for i in $(seq 1 50); do + $CLI GRAPH.QUERY bench "MATCH (a:Node)-[:RELATED_TO]->(b:Node) RETURN a.id, b.id LIMIT 10" > /dev/null 2>&1 && QUERY_OK=$((QUERY_OK + 1)) + done + else + for i in $(seq 1 50); do + $CLI GRAPH.QUERY bench "MATCH (n:Concept) RETURN n LIMIT 10" > /dev/null 2>&1 && QUERY_OK=$((QUERY_OK + 1)) + done + fi + + END_NS=$(date +%s%N) + ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) + QUERY_OPS=$(( 50 * 1000 / (ELAPSED_MS + 1) )) + AVG_US=$(( ELAPSED_MS * 1000 / 50 )) + log " [$ENGINE] 50 Cypher queries in ${ELAPSED_MS}ms (${QUERY_OPS} qps, avg ${AVG_US}µs, $QUERY_OK ok)" + eval "${ENGINE}_CYPHER_MS=$ELAPSED_MS" + eval "${ENGINE}_CYPHER_OPS=$QUERY_OPS" +} + +############################################################################### +# Create graphs +############################################################################### +log "Creating graphs..." +$CLI_MOON GRAPH.CREATE bench > /dev/null 2>&1 +if [[ "$MOON_ONLY" == "false" ]]; then + # FalkorDB auto-creates on first GRAPH.QUERY + $CLI_FALKOR GRAPH.QUERY bench "RETURN 1" > /dev/null 2>&1 +fi + +############################################################################### +# Run benchmarks +############################################################################### +bench_engine "MOON" "$CLI_MOON" "$PORT_MOON" "native" + +if [[ "$MOON_ONLY" == "false" ]]; then + bench_engine "FALKOR" "$CLI_FALKOR" "$PORT_FALKOR" "cypher" +fi + +############################################################################### +# Summary +############################################################################### +echo "" +echo "============================================================" +echo " GRAPH ENGINE BENCHMARK: Moon vs FalkorDB" +echo "============================================================" +echo " Scale: $NODES nodes, $EDGES edges" +echo " Protocol: redis-cli over TCP (sequential, 1 client)" +echo "============================================================" +echo "" + +if [[ "$MOON_ONLY" == "false" ]]; then + printf "%-25s %12s %12s %10s\n" "Operation" "Moon" "FalkorDB" "Ratio" + printf "%-25s %12s %12s %10s\n" "-------------------------" "------------" "------------" "----------" + printf "%-25s %10s/s %10s/s %8.1fx\n" "Node Insert" "$MOON_NODE_OPS" "$FALKOR_NODE_OPS" "$(echo "scale=1; $MOON_NODE_OPS / ($FALKOR_NODE_OPS + 1)" | bc)" + printf "%-25s %10s/s %10s/s %8.1fx\n" "Edge Insert" "$MOON_EDGE_OPS" "$FALKOR_EDGE_OPS" "$(echo "scale=1; $MOON_EDGE_OPS / ($FALKOR_EDGE_OPS + 1)" | bc)" + printf "%-25s %10s/s %10s/s %8.1fx\n" "1-Hop Query" "$MOON_QUERY_OPS" "$FALKOR_QUERY_OPS" "$(echo "scale=1; $MOON_QUERY_OPS / ($FALKOR_QUERY_OPS + 1)" | bc)" + printf "%-25s %10s/s %10s/s %8.1fx\n" "2-Hop Query" "$MOON_2HOP_OPS" "$FALKOR_2HOP_OPS" "$(echo "scale=1; $MOON_2HOP_OPS / ($FALKOR_2HOP_OPS + 1)" | bc)" + printf "%-25s %10s/s %10s/s %8.1fx\n" "Cypher Query" "$MOON_CYPHER_OPS" "$FALKOR_CYPHER_OPS" "$(echo "scale=1; $MOON_CYPHER_OPS / ($FALKOR_CYPHER_OPS + 1)" | bc)" +else + printf "%-25s %12s\n" "Operation" "Moon" + printf "%-25s %12s\n" "-------------------------" "------------" + printf "%-25s %10s/s\n" "Node Insert" "$MOON_NODE_OPS" + printf "%-25s %10s/s\n" "Edge Insert" "$MOON_EDGE_OPS" + printf "%-25s %10s/s\n" "1-Hop Query" "$MOON_QUERY_OPS" + printf "%-25s %10s/s\n" "2-Hop Query" "$MOON_2HOP_OPS" + printf "%-25s %10s/s\n" "Cypher Query" "$MOON_CYPHER_OPS" + echo "" + echo " (FalkorDB comparison skipped — use --moon-only=false with Docker)" +fi + +echo "" +echo "============================================================" +echo " Note: Sequential redis-cli (1 process per command)." +echo " Real throughput is 100-1000x higher with pipelining." +echo " Criterion micro-benchmarks: CSR 1-hop = 923ps, insert = 44ns" +echo "============================================================" diff --git a/scripts/bench-graph-vs-falkor.sh b/scripts/bench-graph-vs-falkor.sh new file mode 100755 index 00000000..add74d5c --- /dev/null +++ b/scripts/bench-graph-vs-falkor.sh @@ -0,0 +1,167 @@ +#!/usr/bin/env bash +# Moon vs FalkorDB — sequential redis-cli, small scale for speed. +set -uo pipefail + +PORT_MOON=16700 +PORT_FALKOR=16701 +NODES=200 +EDGES=600 +QUERIES=100 + +cleanup() { + set +e + [ -n "${MOON_PID:-}" ] && kill "$MOON_PID" 2>/dev/null && wait "$MOON_PID" 2>/dev/null + docker stop falkor_bench 2>/dev/null + set -e +} +trap cleanup EXIT + +log() { printf "[%s] %s\n" "$(date +%H:%M:%S)" "$*"; } +time_ms() { python3 -c "import time; print(int(time.time()*1000))"; } + +# --- Start Moon --- +log "Starting Moon..." +MOON_NO_URING=1 ./target/release/moon --port $PORT_MOON --shards 1 --protected-mode no > /dev/null 2>&1 & +MOON_PID=$! +for i in $(seq 1 30); do redis-cli -p $PORT_MOON PING > /dev/null 2>&1 && break; sleep 0.2; done +log "Moon ready" + +# --- Start FalkorDB --- +log "Starting FalkorDB..." +FALKOR_OK=false +docker run -d --rm --name falkor_bench -p $PORT_FALKOR:6379 falkordb/falkordb:latest > /dev/null 2>&1 || true +for i in $(seq 1 40); do redis-cli -p $PORT_FALKOR PING > /dev/null 2>&1 && { FALKOR_OK=true; break; }; sleep 0.5; done +[ "$FALKOR_OK" = true ] && log "FalkorDB ready" || log "FalkorDB unavailable" + +# ====== MOON BENCHMARK ====== +redis-cli -p $PORT_MOON GRAPH.CREATE bench > /dev/null 2>&1 +log "=== Moon: $NODES nodes ===" +NODE_IDS=() +T0=$(time_ms) +for i in $(seq 1 $NODES); do + L="Concept"; case $((i%5)) in 1)L="Fact";;2)L="Event";;3)L="Source";;4)L="Agent";;esac + R=$(redis-cli -p $PORT_MOON GRAPH.ADDNODE bench "$L" name "k_$i" conf "0.4" 2>/dev/null) + NODE_IDS+=("$R") +done +T1=$(time_ms) +MOON_NODE_MS=$((T1-T0)) +MOON_NODE_OPS=$((NODES*1000/(MOON_NODE_MS+1))) +log " nodes: ${MOON_NODE_MS}ms (${MOON_NODE_OPS}/s)" + +log "=== Moon: $EDGES edges ===" +T0=$(time_ms) +EOK=0 +for i in $(seq 1 $EDGES); do + S=$((i%${#NODE_IDS[@]})); D=$(((i*7+3)%${#NODE_IDS[@]})) + [ $S -eq $D ] && D=$(((D+1)%${#NODE_IDS[@]})) + E="RELATED_TO"; case $((i%5)) in 1)E="DERIVED";;2)E="OBSERVED";;3)E="SUPERSEDES";;4)E="CITED";;esac + redis-cli -p $PORT_MOON GRAPH.ADDEDGE bench "${NODE_IDS[$S]}" "${NODE_IDS[$D]}" "$E" WEIGHT "0.4" > /dev/null 2>&1 && EOK=$((EOK+1)) +done +T1=$(time_ms) +MOON_EDGE_MS=$((T1-T0)) +MOON_EDGE_OPS=$((EDGES*1000/(MOON_EDGE_MS+1))) +log " edges: ${MOON_EDGE_MS}ms (${MOON_EDGE_OPS}/s) [$EOK ok]" + +log "=== Moon: $QUERIES 1-hop ===" +T0=$(time_ms) +for i in $(seq 1 $QUERIES); do + IDX=$((RANDOM%${#NODE_IDS[@]})) + redis-cli -p $PORT_MOON GRAPH.NEIGHBORS bench "${NODE_IDS[$IDX]}" > /dev/null 2>&1 +done +T1=$(time_ms) +MOON_1HOP_MS=$((T1-T0)) +MOON_1HOP_OPS=$((QUERIES*1000/(MOON_1HOP_MS+1))) +log " 1-hop: ${MOON_1HOP_MS}ms (${MOON_1HOP_OPS} qps)" + +log "=== Moon: $QUERIES 2-hop ===" +T0=$(time_ms) +for i in $(seq 1 $QUERIES); do + IDX=$((RANDOM%${#NODE_IDS[@]})) + redis-cli -p $PORT_MOON GRAPH.NEIGHBORS bench "${NODE_IDS[$IDX]}" DEPTH 2 > /dev/null 2>&1 +done +T1=$(time_ms) +MOON_2HOP_MS=$((T1-T0)) +MOON_2HOP_OPS=$((QUERIES*1000/(MOON_2HOP_MS+1))) +log " 2-hop: ${MOON_2HOP_MS}ms (${MOON_2HOP_OPS} qps)" + +# ====== FALKORDB BENCHMARK ====== +FALKOR_NODE_OPS=0; FALKOR_EDGE_OPS=0; FALKOR_1HOP_OPS=0; FALKOR_2HOP_OPS=0 + +if [ "$FALKOR_OK" = true ]; then + redis-cli -p $PORT_FALKOR GRAPH.QUERY bench "RETURN 1" > /dev/null 2>&1 + + log "=== FalkorDB: $NODES nodes ===" + T0=$(time_ms) + for i in $(seq 1 $NODES); do + redis-cli -p $PORT_FALKOR GRAPH.QUERY bench "CREATE (:Node {id:$i,name:'k_$i',conf:0.4})" > /dev/null 2>&1 + done + T1=$(time_ms) + FALKOR_NODE_MS=$((T1-T0)) + FALKOR_NODE_OPS=$((NODES*1000/(FALKOR_NODE_MS+1))) + log " nodes: ${FALKOR_NODE_MS}ms (${FALKOR_NODE_OPS}/s)" + + log "=== FalkorDB: $EDGES edges ===" + T0=$(time_ms) + EOK=0 + for i in $(seq 1 $EDGES); do + S=$(((i%NODES)+1)); D=$((((i*7+3)%NODES)+1)) + [ $S -eq $D ] && D=$(((D%NODES)+1)) + E="RELATED_TO"; case $((i%5)) in 1)E="DERIVED";;2)E="OBSERVED";;3)E="SUPERSEDES";;4)E="CITED";;esac + redis-cli -p $PORT_FALKOR GRAPH.QUERY bench "MATCH (a:Node{id:$S}),(b:Node{id:$D}) CREATE (a)-[:${E}{w:0.4}]->(b)" > /dev/null 2>&1 && EOK=$((EOK+1)) + done + T1=$(time_ms) + FALKOR_EDGE_MS=$((T1-T0)) + FALKOR_EDGE_OPS=$((EDGES*1000/(FALKOR_EDGE_MS+1))) + log " edges: ${FALKOR_EDGE_MS}ms (${FALKOR_EDGE_OPS}/s) [$EOK ok]" + + log "=== FalkorDB: $QUERIES 1-hop ===" + T0=$(time_ms) + for i in $(seq 1 $QUERIES); do + NID=$(((RANDOM%NODES)+1)) + redis-cli -p $PORT_FALKOR GRAPH.QUERY bench "MATCH (a:Node{id:$NID})-[r]->(b) RETURN b.id LIMIT 50" > /dev/null 2>&1 + done + T1=$(time_ms) + FALKOR_1HOP_MS=$((T1-T0)) + FALKOR_1HOP_OPS=$((QUERIES*1000/(FALKOR_1HOP_MS+1))) + log " 1-hop: ${FALKOR_1HOP_MS}ms (${FALKOR_1HOP_OPS} qps)" + + log "=== FalkorDB: $QUERIES 2-hop ===" + T0=$(time_ms) + for i in $(seq 1 $QUERIES); do + NID=$(((RANDOM%NODES)+1)) + redis-cli -p $PORT_FALKOR GRAPH.QUERY bench "MATCH (a:Node{id:$NID})-[*1..2]->(b) RETURN DISTINCT b.id LIMIT 100" > /dev/null 2>&1 + done + T1=$(time_ms) + FALKOR_2HOP_MS=$((T1-T0)) + FALKOR_2HOP_OPS=$((QUERIES*1000/(FALKOR_2HOP_MS+1))) + log " 2-hop: ${FALKOR_2HOP_MS}ms (${FALKOR_2HOP_OPS} qps)" +fi + +# ====== RESULTS ====== +echo "" +echo "============================================================" +echo " Moon vs FalkorDB ($NODES nodes, $EDGES edges, $QUERIES q)" +echo " Sequential redis-cli (fork per command, no pipelining)" +echo "============================================================" +echo "" + +if [ "$FALKOR_OK" = true ]; then + ratio() { python3 -c "print(f'{$1/max($2,1):.1f}')"; } + printf "%-20s %10s %10s %8s\n" "Operation" "Moon" "FalkorDB" "Ratio" + printf "%-20s %10s %10s %8s\n" "---" "---" "---" "---" + printf "%-20s %8s/s %8s/s %6sx\n" "Node Insert" "$MOON_NODE_OPS" "$FALKOR_NODE_OPS" "$(ratio $MOON_NODE_OPS $FALKOR_NODE_OPS)" + printf "%-20s %8s/s %8s/s %6sx\n" "Edge Insert" "$MOON_EDGE_OPS" "$FALKOR_EDGE_OPS" "$(ratio $MOON_EDGE_OPS $FALKOR_EDGE_OPS)" + printf "%-20s %8s/s %8s/s %6sx\n" "1-Hop Query" "$MOON_1HOP_OPS" "$FALKOR_1HOP_OPS" "$(ratio $MOON_1HOP_OPS $FALKOR_1HOP_OPS)" + printf "%-20s %8s/s %8s/s %6sx\n" "2-Hop Query" "$MOON_2HOP_OPS" "$FALKOR_2HOP_OPS" "$(ratio $MOON_2HOP_OPS $FALKOR_2HOP_OPS)" +else + printf "%-20s %10s\n" "Operation" "Moon" + printf "%-20s %10s\n" "---" "---" + printf "%-20s %8s/s\n" "Node Insert" "$MOON_NODE_OPS" + printf "%-20s %8s/s\n" "Edge Insert" "$MOON_EDGE_OPS" + printf "%-20s %8s/s\n" "1-Hop Query" "$MOON_1HOP_OPS" + printf "%-20s %8s/s\n" "2-Hop Query" "$MOON_2HOP_OPS" + echo "(FalkorDB unavailable)" +fi +echo "" +echo "Note: Sequential redis-cli dominates latency (fork+exec per cmd)." +echo "Criterion micro-benchmarks show raw engine: CSR 1-hop=1.02ns, BFS 2-hop=4.99µs" diff --git a/scripts/bench-graph.sh b/scripts/bench-graph.sh new file mode 100755 index 00000000..50132b03 --- /dev/null +++ b/scripts/bench-graph.sh @@ -0,0 +1,265 @@ +#!/usr/bin/env bash +set -euo pipefail + +############################################################################### +# bench-graph.sh -- Graph Engine E2E benchmark with realistic AI agent data +# +# Usage: +# ./scripts/bench-graph.sh # Full run (1K nodes, 5K edges) +# ./scripts/bench-graph.sh --nodes 10000 # Custom node count +# ./scripts/bench-graph.sh --edges 50000 # Custom edge count +# ./scripts/bench-graph.sh --skip-build # Skip cargo build +############################################################################### + +PORT=6501 +NODES=1000 +EDGES=5000 +SHARDS=1 +SKIP_BUILD=false +BINARY="./target/release/moon" +MOON_PID="" + +while [[ $# -gt 0 ]]; do + case "$1" in + --nodes) NODES="$2"; shift 2 ;; + --edges) EDGES="$2"; shift 2 ;; + --shards) SHARDS="$2"; shift 2 ;; + --port) PORT="$2"; shift 2 ;; + --skip-build) SKIP_BUILD=true; shift ;; + *) echo "Unknown: $1"; exit 1 ;; + esac +done + +log() { echo "[$(date '+%H:%M:%S')] $*"; } +err() { echo "[$(date '+%H:%M:%S')] ERROR: $*" >&2; } + +cleanup() { + set +e + if [[ -n "${MOON_PID:-}" ]]; then + kill "$MOON_PID" 2>/dev/null + wait "$MOON_PID" 2>/dev/null + fi + set -e +} +trap cleanup EXIT + +CLI="redis-cli -p $PORT" + +############################################################################### +# 1. Build +############################################################################### +if [[ "$SKIP_BUILD" == "false" ]]; then + log "Building Moon with graph feature..." + cargo build --release --no-default-features --features runtime-tokio,jemalloc,graph 2>&1 | tail -1 +fi + +############################################################################### +# 2. Start Moon +############################################################################### +log "Starting Moon on port $PORT (shards=$SHARDS)..." +MOON_NO_URING=1 $BINARY --port $PORT --shards $SHARDS --protected-mode no > /tmp/moon_graph_bench.log 2>&1 & +MOON_PID=$! + +# Wait for server to be ready (poll up to 10 seconds) +for attempt in $(seq 1 20); do + if $CLI PING > /dev/null 2>&1; then + break + fi + sleep 0.5 +done + +if ! $CLI PING > /dev/null 2>&1; then + err "Moon failed to start after 10s. Log:" + cat /tmp/moon_graph_bench.log + exit 1 +fi +log "Moon started (PID=$MOON_PID)" + +############################################################################### +# 3. AI Agent Knowledge Graph — Realistic Mock Data +############################################################################### +GRAPH="{agent-bench}:knowledge" + +log "=== Phase 1: Graph Creation ===" +set +e +RESULT=$($CLI GRAPH.CREATE "$GRAPH" 2>&1) +RC=$? +set -e +echo " GRAPH.CREATE: $RESULT (rc=$RC)" +if [[ "$RESULT" == *"ERR"* ]]; then + err "Graph creation failed. Aborting." + exit 1 +fi + +log "=== Phase 2: Node Insertion ($NODES nodes) ===" +NODE_IDS=() +LABELS=("Concept" "Fact" "Event" "Source" "Agent") +START_NS=$(date +%s%N) + +for i in $(seq 1 $NODES); do + LABEL=${LABELS[$((i % 5))]} + set +e + RESULT=$($CLI GRAPH.ADDNODE "$GRAPH" "$LABEL" name "knowledge_$i" confidence "0.42" created_at "$((1712800000 + i))" 2>&1) + set -e + if [[ "$RESULT" == *"ERR"* ]]; then + if [[ $i -le 3 ]]; then err " ADDNODE $i failed: $RESULT"; fi + else + NODE_IDS+=("$RESULT") + fi +done +ACTUAL_NODES=${#NODE_IDS[@]} +if [[ $ACTUAL_NODES -eq 0 ]]; then + err "No nodes inserted. First result was: $RESULT" + exit 1 +fi + +END_NS=$(date +%s%N) +ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) +OPS_PER_SEC=$(( NODES * 1000 / (ELAPSED_MS + 1) )) +log " $NODES nodes inserted in ${ELAPSED_MS}ms (${OPS_PER_SEC} ops/s)" + +log "=== Phase 3: Edge Insertion ($EDGES edges) ===" +EDGE_TYPES=("RELATED_TO" "DERIVED_FROM" "OBSERVED_AT" "SUPERSEDES" "CITED_BY") +START_NS=$(date +%s%N) + +EDGE_OK=0 +EDGE_ERR=0 +for i in $(seq 1 $EDGES); do + SRC_IDX=$(( RANDOM % ACTUAL_NODES )) + DST_IDX=$(( RANDOM % ACTUAL_NODES )) + # Avoid self-loops + if [[ $DST_IDX -eq $SRC_IDX ]]; then + DST_IDX=$(( (SRC_IDX + 1) % ACTUAL_NODES )) + fi + SRC_ID="${NODE_IDS[$SRC_IDX]}" + DST_ID="${NODE_IDS[$DST_IDX]}" + ETYPE=${EDGE_TYPES[$((i % 5))]} + set +e + RESULT=$($CLI GRAPH.ADDEDGE "$GRAPH" "$SRC_ID" "$DST_ID" "$ETYPE" WEIGHT "0.42" 2>&1) + set -e + if [[ "$RESULT" == *"ERR"* ]]; then + EDGE_ERR=$((EDGE_ERR + 1)) + if [[ $EDGE_ERR -le 3 ]]; then + err " Edge $i failed: $RESULT (src=$SRC_ID dst=$DST_ID)" + fi + else + EDGE_OK=$((EDGE_OK + 1)) + fi +done +log " $EDGE_OK edges ok, $EDGE_ERR errors" + +END_NS=$(date +%s%N) +ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) +OPS_PER_SEC=$(( EDGES * 1000 / (ELAPSED_MS + 1) )) +log " $EDGES edges inserted in ${ELAPSED_MS}ms (${OPS_PER_SEC} ops/s)" + +log "=== Phase 4: Graph Info ===" +$CLI GRAPH.INFO "$GRAPH" 2>&1 | head -20 + +log "=== Phase 5: 1-Hop Neighbor Queries (100 random) ===" +START_NS=$(date +%s%N) +SUCCESS=0 +EMPTY=0 +for i in $(seq 1 100); do + IDX=$(( RANDOM % NODES )) + NID="${NODE_IDS[$IDX]}" + RESULT=$($CLI GRAPH.NEIGHBORS "$GRAPH" "$NID" 2>&1) + if [[ "$RESULT" != *"ERR"* ]]; then + SUCCESS=$((SUCCESS + 1)) + if [[ "$RESULT" == "(empty"* ]] || [[ "$RESULT" == "" ]]; then + EMPTY=$((EMPTY + 1)) + fi + fi +done +END_NS=$(date +%s%N) +ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) +AVG_US=$(( ELAPSED_MS * 1000 / 100 )) +log " 100 queries in ${ELAPSED_MS}ms (avg ${AVG_US}µs/query, $SUCCESS ok, $EMPTY empty)" + +log "=== Phase 6: 2-Hop Neighbor Queries (50 random) ===" +START_NS=$(date +%s%N) +SUCCESS=0 +for i in $(seq 1 50); do + IDX=$(( RANDOM % NODES )) + NID="${NODE_IDS[$IDX]}" + RESULT=$($CLI GRAPH.NEIGHBORS "$GRAPH" "$NID" DEPTH 2 2>&1) + if [[ "$RESULT" != *"ERR"* ]]; then + SUCCESS=$((SUCCESS + 1)) + fi +done +END_NS=$(date +%s%N) +ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) +AVG_US=$(( ELAPSED_MS * 1000 / 50 )) +log " 50 queries in ${ELAPSED_MS}ms (avg ${AVG_US}µs/query, $SUCCESS ok)" + +log "=== Phase 7: Typed Edge Queries (50 random, TYPE RELATED_TO) ===" +START_NS=$(date +%s%N) +SUCCESS=0 +for i in $(seq 1 50); do + IDX=$(( RANDOM % NODES )) + NID="${NODE_IDS[$IDX]}" + RESULT=$($CLI GRAPH.NEIGHBORS "$GRAPH" "$NID" TYPE RELATED_TO 2>&1) + if [[ "$RESULT" != *"ERR"* ]]; then + SUCCESS=$((SUCCESS + 1)) + fi +done +END_NS=$(date +%s%N) +ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) +AVG_US=$(( ELAPSED_MS * 1000 / 50 )) +log " 50 queries in ${ELAPSED_MS}ms (avg ${AVG_US}µs/query, $SUCCESS ok)" + +log "=== Phase 8: Cypher Query (MATCH + RETURN) ===" +START_NS=$(date +%s%N) +for i in $(seq 1 20); do + RESULT=$($CLI GRAPH.QUERY "$GRAPH" "MATCH (n:Concept) RETURN n LIMIT 10" 2>&1) +done +END_NS=$(date +%s%N) +ELAPSED_MS=$(( (END_NS - START_NS) / 1000000 )) +AVG_US=$(( ELAPSED_MS * 1000 / 20 )) +log " 20 Cypher queries in ${ELAPSED_MS}ms (avg ${AVG_US}µs/query)" + +log "=== Phase 9: GRAPH.EXPLAIN (query plan) ===" +RESULT=$($CLI GRAPH.EXPLAIN "$GRAPH" "MATCH (n:Concept)-[:RELATED_TO]->(m) RETURN n, m LIMIT 10" 2>&1) +echo " Plan: $(echo "$RESULT" | head -5)" + +log "=== Phase 10: Graph List ===" +$CLI GRAPH.LIST 2>&1 + +log "=== Phase 11: KV Regression Check (1000 SET + 1000 GET) ===" +START_NS=$(date +%s%N) +for i in $(seq 1 1000); do + $CLI SET "bench_key_$i" "value_$i" > /dev/null 2>&1 +done +END_NS=$(date +%s%N) +SET_MS=$(( (END_NS - START_NS) / 1000000 )) + +START_NS=$(date +%s%N) +for i in $(seq 1 1000); do + $CLI GET "bench_key_$i" > /dev/null 2>&1 +done +END_NS=$(date +%s%N) +GET_MS=$(( (END_NS - START_NS) / 1000000 )) + +SET_OPS=$(( 1000 * 1000 / (SET_MS + 1) )) +GET_OPS=$(( 1000 * 1000 / (GET_MS + 1) )) +log " SET: 1000 ops in ${SET_MS}ms (${SET_OPS} ops/s)" +log " GET: 1000 ops in ${GET_MS}ms (${GET_OPS} ops/s)" + +log "=== Phase 12: Graph Delete ===" +$CLI GRAPH.DELETE "$GRAPH" 2>&1 +RESULT=$($CLI GRAPH.LIST 2>&1) +echo " After delete: $RESULT" + +############################################################################### +# Summary +############################################################################### +echo "" +echo "============================================================" +echo " GRAPH ENGINE E2E BENCHMARK COMPLETE" +echo "============================================================" +echo " Nodes: $NODES | Edges: $EDGES | Shards: $SHARDS" +echo " Server: Moon with graph feature (tokio runtime)" +echo "============================================================" + +cleanup +log "Done." diff --git a/src/command/connection.rs b/src/command/connection.rs index 3f91e743..ef197fe9 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -271,6 +271,13 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { ); sections.push_str("\r\n"); + // # Commandstats — placeholder section for Redis 7.x parity. + // Per-command stats (calls, usec, usec_per_call) require a global registry; + // the record_command() path already tracks per-label counters in Prometheus. + // For now, emit the section header so redis-py parse_info recognizes it. + sections.push_str("# Commandstats\r\n"); + sections.push_str("\r\n"); + sections.push_str("# Keyspace\r\n"); let key_count = db.len(); let expires_count = db.expires_count(); diff --git a/src/command/graph/graph_read.rs b/src/command/graph/graph_read.rs new file mode 100644 index 00000000..a441ef3b --- /dev/null +++ b/src/command/graph/graph_read.rs @@ -0,0 +1,1369 @@ +//! GRAPH.* read command handlers. +//! +//! These commands read from GraphStore: NEIGHBORS, INFO, LIST, QUERY, RO_QUERY, EXPLAIN. + +use bytes::Bytes; +use slotmap::Key; + +use crate::graph::cypher; +use crate::graph::store::GraphStore; +use crate::graph::traversal::SegmentMergeReader; +use crate::graph::types::Direction; +use crate::protocol::Frame; + +use super::graph_write::extract_bulk; + +/// GRAPH.NEIGHBORS [TYPE ] [DEPTH ] +/// +/// Returns an array of neighbor nodes/edges as RESP3 Maps. +/// Default direction: BOTH (outgoing + incoming). +/// DEPTH > 1 performs multi-hop expansion (BFS). +pub fn graph_neighbors(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.NEIGHBORS' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let node_id = match parse_u64(&args[1]) { + Some(id) => id, + None => return Frame::Error(Bytes::from_static(b"ERR invalid node ID")), + }; + + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + // Parse optional TYPE and DEPTH arguments. + let mut edge_type_filter: Option = None; + let mut depth: u32 = 1; + let mut pos = 2; + + while pos < args.len() { + let key = match extract_bulk(&args[pos]) { + Some(b) => b, + None => { + pos += 1; + continue; + } + }; + + if key.eq_ignore_ascii_case(b"TYPE") { + pos += 1; + if pos >= args.len() { + return Frame::Error(Bytes::from_static(b"ERR missing TYPE value")); + } + if let Some(type_name) = extract_bulk(&args[pos]) { + edge_type_filter = Some(super::graph_write::label_to_id(type_name)); + } + pos += 1; + } else if key.eq_ignore_ascii_case(b"DEPTH") { + pos += 1; + if pos >= args.len() { + return Frame::Error(Bytes::from_static(b"ERR missing DEPTH value")); + } + depth = match parse_u32(&args[pos]) { + Some(d) if d > 0 => d, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid DEPTH value")), + }; + pos += 1; + } else { + pos += 1; + } + } + + // Cap depth to prevent explosion. + let max_depth = 10u32; + if depth > max_depth { + return Frame::Error(Bytes::from_static(b"ERR DEPTH exceeds maximum (10)")); + } + + let node_key = super::graph_write::external_id_to_node_key(node_id); + + let memgraph = &graph.write_buf; + + // Verify node exists in the mutable write buffer. + // Nodes in CSR segments still have MemGraph entries (freeze copies, doesn't move). + if memgraph.get_node(node_key).is_none() { + return Frame::Error(Bytes::from_static(b"ERR node not found")); + } + + // Build a SegmentMergeReader that sees both MemGraph and immutable CSR segments. + let lsn = u64::MAX - 1; // See all live data (MAX-1 because deleted_lsn=MAX means alive). + let segments_guard = graph.segments.load(); + let csr_segs = &segments_guard.immutable; + let reader = SegmentMergeReader::new( + Some(memgraph), + csr_segs, + Direction::Both, + lsn, + edge_type_filter, + ); + + // TraversalGuard enforces bounded epoch hold (30s default timeout). + let guard = crate::graph::traversal_guard::TraversalGuard::with_default_timeout(lsn); + + // BFS expansion using SegmentMergeReader for per-node neighbor lookup. + let mut visited = std::collections::HashSet::new(); + visited.insert(node_id); + let mut frontier = vec![node_key]; + let mut results: Vec = Vec::with_capacity(128); + // Cap total results. + let max_results = 10_000usize; + + for _hop in 0..depth { + // Check traversal timeout at each hop. + if let Err(timeout) = guard.check_timeout() { + return Frame::Error(Bytes::from(format!("ERR {timeout}"))); + } + + let mut next_frontier = Vec::with_capacity(frontier.len() * 4); + + for ¤t in &frontier { + for merged in reader.neighbors(current) { + let neighbor_ext_id = merged.node.data().as_ffi(); + + if visited.contains(&neighbor_ext_id) { + continue; + } + visited.insert(neighbor_ext_id); + + // Add edge as RESP3 Map (from MemGraph if available, otherwise synthetic). + if let Some(edge) = memgraph.get_edge(merged.edge) { + results.push(edge_to_frame(merged.edge, edge)); + } else { + // CSR-only edge: build a minimal edge frame from MergedNeighbor. + results.push(merged_edge_to_frame(&merged)); + } + + // Add neighbor node as RESP3 Map. + if let Some(node) = memgraph.get_node(merged.node) { + results.push(node_to_frame(merged.node, node)); + } else { + // CSR-only node: minimal node frame. + results.push(merged_node_to_frame(&merged)); + } + + if results.len() >= max_results { + break; + } + + next_frontier.push(merged.node); + } + + if results.len() >= max_results { + break; + } + } + + frontier = next_frontier; + if frontier.is_empty() { + break; + } + } + + Frame::Array(results.into()) +} + +/// GRAPH.INFO +/// +/// Returns graph statistics as RESP3 Map. +pub fn graph_info(store: &GraphStore, args: &[Frame]) -> Frame { + if args.is_empty() { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.INFO' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let memgraph = &graph.write_buf; + let segments = graph.segments.load(); + let stats = &graph.stats; + + let node_count = memgraph.node_count() as i64; + let edge_count = memgraph.edge_count() as i64; + let immutable_segments = segments.immutable.len() as i64; + + // Degree distribution from GraphStats. + let degree_stats = Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"avg")), + Frame::Double(stats.degree_stats.avg), + ), + ( + Frame::SimpleString(Bytes::from_static(b"p50")), + Frame::Integer(stats.degree_stats.p50 as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"p99")), + Frame::Integer(stats.degree_stats.p99 as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"max")), + Frame::Integer(stats.degree_stats.max as i64), + ), + ]); + + Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"name")), + Frame::BulkString(graph.name.clone()), + ), + ( + Frame::SimpleString(Bytes::from_static(b"node_count")), + Frame::Integer(node_count), + ), + ( + Frame::SimpleString(Bytes::from_static(b"edge_count")), + Frame::Integer(edge_count), + ), + ( + Frame::SimpleString(Bytes::from_static(b"immutable_segments")), + Frame::Integer(immutable_segments), + ), + ( + Frame::SimpleString(Bytes::from_static(b"edge_threshold")), + Frame::Integer(graph.edge_threshold as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"created_lsn")), + Frame::Integer(graph.created_lsn as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"degree_stats")), + degree_stats, + ), + ]) +} + +/// GRAPH.LIST +/// +/// Returns an array of all graph names. +pub fn graph_list(store: &GraphStore) -> Frame { + let names = store.list_graphs(); + let frames: Vec = names + .into_iter() + .map(|name| Frame::BulkString(name.clone())) + .collect(); + Frame::Array(frames.into()) +} + +// --------------------------------------------------------------------------- +// GRAPH.QUERY, GRAPH.RO_QUERY, GRAPH.EXPLAIN +// --------------------------------------------------------------------------- + +/// GRAPH.QUERY +/// +/// Parses the Cypher query, compiles to a physical plan, executes it against +/// the named graph, and returns result rows with column headers and statistics. +pub fn graph_query(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.QUERY' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let cypher_bytes = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid Cypher query")), + }; + + let query = match cypher::parse_cypher(cypher_bytes) { + Ok(q) => q, + Err(e) => { + let msg = format!("ERR Cypher parse error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + // Plan cache: check for a cached plan before compiling. + let query_hash = cypher::planner::hash_query(cypher_bytes); + let plan = { + let cache = graph.plan_cache.lock(); + cache.get(query_hash) + }; + let plan = if let Some(cached) = plan { + cached + } else { + let p = match cypher::planner::compile(&query) { + Ok(p) => std::sync::Arc::new(p), + Err(e) => { + let msg = format!("ERR Cypher plan error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + graph.plan_cache.lock().insert(query_hash, p.clone()); + p + }; + + let params = std::collections::HashMap::new(); + let result = match cypher::executor::execute(graph, &plan, ¶ms) { + Ok(r) => r, + Err(e) => { + let msg = format!("ERR Cypher execution error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + exec_result_to_frame(&result) +} + +/// GRAPH.QUERY — write-capable variant. +/// +/// Called when the Cypher query contains write clauses (CREATE, DELETE, SET, MERGE). +/// Takes `&mut GraphStore` to allow mutable access to the named graph. +pub fn graph_query_write(store: &mut GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.QUERY' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let cypher_bytes = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid Cypher query")), + }; + + let query = match cypher::parse_cypher(cypher_bytes) { + Ok(q) => q, + Err(e) => { + let msg = format!("ERR Cypher parse error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + let plan = match cypher::planner::compile(&query) { + Ok(p) => p, + Err(e) => { + let msg = format!("ERR Cypher plan error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + let lsn = store.allocate_lsn(); + + // Scoped borrow: get mutable graph, execute, release borrow before WAL push. + let result = { + let graph = match store.get_graph_mut(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let params = std::collections::HashMap::new(); + match cypher::executor::execute_mut(graph, &plan, ¶ms, lsn) { + Ok(r) => r, + Err(e) => { + let msg = format!("ERR Cypher execution error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + } + }; + + // Generate WAL records for mutations performed during execution. + for mutation in &result.mutations { + match mutation { + cypher::executor::MutationRecord::CreateNode { + node_id, + labels, + properties, + embedding, + } => { + store + .wal_pending + .push(crate::graph::wal::serialize_add_node( + graph_name, + *node_id, + labels, + properties, + embedding.as_deref(), + )); + } + cypher::executor::MutationRecord::CreateEdge { + edge_id, + src_id, + dst_id, + edge_type, + weight, + properties, + } => { + store + .wal_pending + .push(crate::graph::wal::serialize_add_edge( + graph_name, + *edge_id, + *src_id, + *dst_id, + *edge_type, + *weight, + properties.as_ref(), + )); + } + } + } + + exec_result_to_frame(&result) +} + +/// GRAPH.QUERY — auto-routing handler. +/// +/// Parses the Cypher query once, then dispatches to the read or write path +/// based on whether the query contains write clauses. +pub fn graph_query_or_write(store: &mut GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.QUERY' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let cypher_bytes = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid Cypher query")), + }; + + // Parse once — no double-parse overhead. + let query = match cypher::parse_cypher(cypher_bytes) { + Ok(q) => q, + Err(e) => { + let msg = format!("ERR Cypher parse error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + if query.is_read_only() { + // Read path: compile plan (with cache), execute read-only. + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let query_hash = cypher::planner::hash_query(cypher_bytes); + let plan = { + let cache = graph.plan_cache.lock(); + cache.get(query_hash) + }; + let plan = if let Some(cached) = plan { + cached + } else { + let p = match cypher::planner::compile(&query) { + Ok(p) => std::sync::Arc::new(p), + Err(e) => { + let msg = format!("ERR Cypher plan error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + graph.plan_cache.lock().insert(query_hash, p.clone()); + p + }; + + let params = std::collections::HashMap::new(); + let result = match cypher::executor::execute(graph, &plan, ¶ms) { + Ok(r) => r, + Err(e) => { + let msg = format!("ERR Cypher execution error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + exec_result_to_frame(&result) + } else { + // Write path: compile plan (no cache for writes), execute with mutations. + let plan = match cypher::planner::compile(&query) { + Ok(p) => p, + Err(e) => { + let msg = format!("ERR Cypher plan error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + let lsn = store.allocate_lsn(); + + let result = { + let graph = match store.get_graph_mut(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let params = std::collections::HashMap::new(); + match cypher::executor::execute_mut(graph, &plan, ¶ms, lsn) { + Ok(r) => r, + Err(e) => { + let msg = format!("ERR Cypher execution error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + } + }; + + // Generate WAL records for mutations. + for mutation in &result.mutations { + match mutation { + cypher::executor::MutationRecord::CreateNode { + node_id, + labels, + properties, + embedding, + } => { + store + .wal_pending + .push(crate::graph::wal::serialize_add_node( + graph_name, + *node_id, + labels, + properties, + embedding.as_deref(), + )); + } + cypher::executor::MutationRecord::CreateEdge { + edge_id, + src_id, + dst_id, + edge_type, + weight, + properties, + } => { + store + .wal_pending + .push(crate::graph::wal::serialize_add_edge( + graph_name, + *edge_id, + *src_id, + *dst_id, + *edge_type, + *weight, + properties.as_ref(), + )); + } + } + } + + exec_result_to_frame(&result) + } +} + +/// GRAPH.RO_QUERY +/// +/// Like GRAPH.QUERY but rejects write clauses (CREATE, DELETE, SET, MERGE). +pub fn graph_ro_query(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.RO_QUERY' command", + )); + } + + let cypher_bytes = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid Cypher query")), + }; + + let query = match cypher::parse_cypher(cypher_bytes) { + Ok(q) => q, + Err(e) => { + let msg = format!("ERR Cypher parse error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + if !query.is_read_only() { + return Frame::Error(Bytes::from_static( + b"ERR GRAPH.RO_QUERY does not allow write clauses (CREATE, DELETE, SET, MERGE)", + )); + } + + // Delegate to the regular query handler for parsing/planning. + graph_query(store, args) +} + +/// GRAPH.EXPLAIN +/// +/// Returns the execution plan without running the query. +/// Includes cost-based strategy selection when graph stats are available. +pub fn graph_explain(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.EXPLAIN' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let cypher_bytes = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid Cypher query")), + }; + + let query = match cypher::parse_cypher(cypher_bytes) { + Ok(q) => q, + Err(e) => { + let msg = format!("ERR Cypher parse error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + let plan = match cypher::planner::compile(&query) { + Ok(p) => p, + Err(e) => { + let msg = format!("ERR Cypher plan error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + // Return plan as a formatted string. + let mut output = String::new(); + for (i, op) in plan.operators.iter().enumerate() { + if !output.is_empty() { + output.push('\n'); + } + output.push_str(&format!("{i}: {op:?}")); + } + + // Append cost-based strategy selection if graph exists. + if let Some(graph) = store.get_graph(graph_name) { + let stats = &graph.stats; + + // Extract traversal parameters from the plan operators. + let hops = extract_max_hops(&plan); + let k = 10u32; // Default k for vector search. + let dim = 128u32; // Default dimension estimate. + + let estimate = cypher::planner::select_strategy( + stats, 1, // start_nodes (single seed) + hops, k, dim, None, // No specific start node degree without node ID. + ); + + output.push_str(&format!( + "\n--- Cost Estimation ---\nStrategy: {}\nGraph-first cost: {:.1}\nVector-first cost: {:.1}\nHub detected: {}", + estimate.strategy, + estimate.graph_first_cost, + estimate.vector_first_cost, + estimate.hub_detected, + )); + } + + Frame::BulkString(Bytes::from(output)) +} + +/// GRAPH.PROFILE +/// +/// Execute query with per-operator timing. Returns `[results, operator_profiles]`. +pub fn graph_profile(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 2 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.PROFILE' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let cypher_bytes = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid Cypher query")), + }; + + let query = match cypher::parse_cypher(cypher_bytes) { + Ok(q) => q, + Err(e) => { + let msg = format!("ERR Cypher parse error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + let plan = match cypher::planner::compile(&query) { + Ok(p) => p, + Err(e) => { + let msg = format!("ERR Cypher plan error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + let params = std::collections::HashMap::new(); + let profile = match cypher::executor::execute_profile(graph, &plan, ¶ms) { + Ok(r) => r, + Err(e) => { + let msg = format!("ERR Cypher execution error: {e}"); + return Frame::Error(Bytes::from(msg)); + } + }; + + profile_result_to_frame(&profile) +} + +/// Convert a `ProfileResult` to a RESP3 Frame. +/// +/// Format: Array [ +/// exec_result_frame, // same format as GRAPH.QUERY +/// Array [ // per-operator profiles +/// Array [name, row_count, duration_us], +/// ... +/// ] +/// ] +fn profile_result_to_frame(profile: &cypher::executor::ProfileResult) -> Frame { + let result_frame = exec_result_to_frame(&profile.exec_result); + + let op_frames: Vec = profile + .operator_profiles + .iter() + .map(|op| { + Frame::Array( + vec![ + Frame::BulkString(Bytes::from(op.name)), + Frame::Integer(op.row_count as i64), + Frame::Integer(op.duration_us as i64), + ] + .into(), + ) + }) + .collect(); + + Frame::Array(vec![result_frame, Frame::Array(op_frames.into())].into()) +} + +/// Convert an `ExecResult` to a RESP3 Frame. +/// +/// Format: Array [ +/// Array [column_name_1, column_name_2, ...], // headers +/// Array [ Array [val1, val2, ...], ... ], // rows +/// BulkString "Nodes created: N, ..." // stats +/// ] +fn exec_result_to_frame(result: &cypher::executor::ExecResult) -> Frame { + // 1. Headers + let headers: Vec = result + .columns + .iter() + .map(|c| Frame::BulkString(Bytes::from(c.clone()))) + .collect(); + + // 2. Rows -- each row is an array of values. + let rows: Vec = result + .rows + .iter() + .map(|row| { + let cells: Vec = row.iter().map(value_to_frame).collect(); + Frame::Array(cells.into()) + }) + .collect(); + + // 3. Stats — use write! to pre-allocated buffer instead of format!() + let mut stats_buf = Vec::with_capacity(128); + use std::io::Write as _; + let _ = write!( + stats_buf, + "Nodes created: {}, Nodes deleted: {}, Properties set: {}, \ + Query internal execution time: {} us", + result.nodes_created, result.nodes_deleted, result.properties_set, result.execution_time_us + ); + + Frame::Array( + vec![ + Frame::Array(headers.into()), + Frame::Array(rows.into()), + Frame::BulkString(Bytes::from(stats_buf)), + ] + .into(), + ) +} + +/// Convert an executor Value to a Frame. +/// +/// Uses `itoa` + stack buffer for Node/Edge IDs to avoid per-row `format!()` +/// heap allocations on the query result serialization hot path. +fn value_to_frame(value: &cypher::executor::Value) -> Frame { + use cypher::executor::Value; + match value { + Value::Null => Frame::Null, + Value::Int(n) => Frame::Integer(*n), + Value::Float(f) => Frame::Double(*f), + Value::String(s) => Frame::BulkString(Bytes::from(s.clone())), + Value::Bool(b) => Frame::Boolean(*b), + Value::Node(key) => { + // "node:" (5) + max u64 (20 digits) = 25 bytes max + let mut buf = [0u8; 32]; + buf[..5].copy_from_slice(b"node:"); + let mut itoa_buf = itoa::Buffer::new(); + let n = itoa_buf.format(key.data().as_ffi()); + let end = 5 + n.len(); + buf[5..end].copy_from_slice(n.as_bytes()); + Frame::BulkString(Bytes::copy_from_slice(&buf[..end])) + } + Value::Edge(key) => { + let mut buf = [0u8; 32]; + buf[..5].copy_from_slice(b"edge:"); + let mut itoa_buf = itoa::Buffer::new(); + let n = itoa_buf.format(key.data().as_ffi()); + let end = 5 + n.len(); + buf[5..end].copy_from_slice(n.as_bytes()); + Frame::BulkString(Bytes::copy_from_slice(&buf[..end])) + } + Value::List(items) => { + let frames: Vec = items.iter().map(value_to_frame).collect(); + Frame::Array(frames.into()) + } + Value::Map(entries) => { + let pairs: Vec<(Frame, Frame)> = entries + .iter() + .map(|(k, v)| (Frame::BulkString(Bytes::from(k.clone())), value_to_frame(v))) + .collect(); + Frame::Map(pairs) + } + } +} + +/// Extract the maximum hop count from Expand operators in a physical plan. +fn extract_max_hops(plan: &cypher::planner::PhysicalPlan) -> u32 { + let mut max_hops = 1u32; + for op in &plan.operators { + if let cypher::planner::PhysicalOp::Expand { max_hops: mh, .. } = op { + if *mh > max_hops { + max_hops = *mh; + } + } + } + max_hops +} + +// --------------------------------------------------------------------------- +// RESP3 entity formatting +// --------------------------------------------------------------------------- + +/// Format a node as a RESP3 Map: {id, labels, properties}. +/// Format a CSR-only edge from a MergedNeighbor as a RESP3 Map. +/// Used when the edge exists only in immutable CSR segments. +fn merged_edge_to_frame(merged: &crate::graph::traversal::MergedNeighbor) -> Frame { + Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"id")), + Frame::Integer(merged.edge.data().as_ffi() as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"type")), + Frame::Integer(merged.edge_type as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"weight")), + Frame::Double(merged.weight), + ), + ( + Frame::SimpleString(Bytes::from_static(b"properties")), + Frame::Map(Vec::new()), + ), + ]) +} + +/// Format a CSR-only node from a MergedNeighbor as a RESP3 Map. +/// Used when the node exists only in immutable CSR segments. +fn merged_node_to_frame(merged: &crate::graph::traversal::MergedNeighbor) -> Frame { + let external_id = merged.node.data().as_ffi(); + Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"id")), + Frame::Integer(external_id as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"labels")), + Frame::Array(Vec::new().into()), + ), + ( + Frame::SimpleString(Bytes::from_static(b"properties")), + Frame::Map(Vec::new()), + ), + ]) +} + +fn node_to_frame( + key: crate::graph::types::NodeKey, + node: &crate::graph::types::MutableNode, +) -> Frame { + let external_id = key.data().as_ffi(); + + let labels: Vec = node + .labels + .iter() + .map(|&l| Frame::Integer(l as i64)) + .collect(); + + let props = properties_to_frame(&node.properties); + + Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"id")), + Frame::Integer(external_id as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"labels")), + Frame::Array(labels.into()), + ), + ( + Frame::SimpleString(Bytes::from_static(b"properties")), + props, + ), + ]) +} + +/// Format an edge as a RESP3 Map: {id, type, src, dst, properties}. +fn edge_to_frame( + key: crate::graph::types::EdgeKey, + edge: &crate::graph::types::MutableEdge, +) -> Frame { + let external_id = key.data().as_ffi(); + + let src_ext = edge.src.data().as_ffi(); + let dst_ext = edge.dst.data().as_ffi(); + + let props = match &edge.properties { + Some(p) => properties_to_frame(p), + None => Frame::Map(Vec::new()), + }; + + Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"id")), + Frame::Integer(external_id as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"type")), + Frame::Integer(edge.edge_type as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"src")), + Frame::Integer(src_ext as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"dst")), + Frame::Integer(dst_ext as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"weight")), + Frame::Double(edge.weight), + ), + ( + Frame::SimpleString(Bytes::from_static(b"properties")), + props, + ), + ]) +} + +/// Convert a PropertyMap to a RESP3 Map frame. +fn properties_to_frame(props: &crate::graph::types::PropertyMap) -> Frame { + let pairs: Vec<(Frame, Frame)> = props + .iter() + .map(|(key, val)| { + let k = Frame::Integer(*key as i64); + let v = match val { + crate::graph::types::PropertyValue::Int(n) => Frame::Integer(*n), + crate::graph::types::PropertyValue::Float(f) => Frame::Double(*f), + crate::graph::types::PropertyValue::String(s) => Frame::BulkString(s.clone()), + crate::graph::types::PropertyValue::Bool(b) => Frame::Boolean(*b), + crate::graph::types::PropertyValue::Bytes(b) => Frame::BulkString(b.clone()), + }; + (k, v) + }) + .collect(); + Frame::Map(pairs) +} + +// --------------------------------------------------------------------------- +// GRAPH.VSEARCH — graph-filtered vector search (HYB-01) +// --------------------------------------------------------------------------- + +/// GRAPH.VSEARCH [THRESHOLD ] [TYPE ] +/// +/// Traverses `hops` from `start_node_id`, collects candidate nodes, then scores +/// by cosine similarity to `vector_blob`. Returns top-K results. +pub fn graph_vsearch(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 5 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.VSEARCH' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let start_id = match parse_u64(&args[1]) { + Some(id) => id, + None => return Frame::Error(Bytes::from_static(b"ERR invalid start node ID")), + }; + + let hops = match parse_u32(&args[2]) { + Some(h) if h > 0 && h <= 10 => h, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid hops (1-10)")), + }; + + let k = match parse_u32(&args[3]) { + Some(k) if k > 0 => k as usize, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid k")), + }; + + let query_vector = match extract_f32_vector(&args[4]) { + Some(v) if !v.is_empty() => v, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid vector blob")), + }; + + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + // Parse optional args. + let mut threshold = crate::graph::hybrid::DEFAULT_STRATEGY_THRESHOLD; + let mut edge_type_filter: Option = None; + let mut pos = 5; + while pos < args.len() { + let key = match extract_bulk(&args[pos]) { + Some(b) => b, + None => { + pos += 1; + continue; + } + }; + if key.eq_ignore_ascii_case(b"THRESHOLD") { + pos += 1; + if pos < args.len() { + if let Some(t) = parse_u32(&args[pos]) { + threshold = t as usize; + } + } + pos += 1; + } else if key.eq_ignore_ascii_case(b"TYPE") { + pos += 1; + if pos < args.len() { + if let Some(type_name) = extract_bulk(&args[pos]) { + edge_type_filter = Some(super::graph_write::label_to_id(type_name)); + } + } + pos += 1; + } else { + pos += 1; + } + } + + let node_key = super::graph_write::external_id_to_node_key(start_id); + let memgraph = &graph.write_buf; + let lsn = u64::MAX - 1; + + let mut search = + crate::graph::hybrid::GraphFilteredSearch::new(node_key, hops, query_vector, k); + search.threshold = threshold; + search.edge_type_filter = edge_type_filter; + + match search.execute(memgraph, lsn) { + Ok(results) => hybrid_results_to_frame(&results), + Err(e) => Frame::Error(Bytes::from(format!("ERR {e}"))), + } +} + +// --------------------------------------------------------------------------- +// GRAPH.HYBRID — general hybrid query dispatcher +// --------------------------------------------------------------------------- + +/// GRAPH.HYBRID +/// +/// Modes: +/// FILTER — graph-filtered vector search (HYB-01) +/// EXPAND — vector-to-graph expansion (HYB-02) +/// WALK — vector-guided walk (HYB-03) +/// RERANK — graph-constrained re-ranking (HYB-04) +pub fn graph_hybrid(store: &GraphStore, args: &[Frame]) -> Frame { + if args.len() < 3 { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.HYBRID' command", + )); + } + + let graph_name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + let mode = match extract_bulk(&args[1]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid mode")), + }; + + let graph = match store.get_graph(graph_name) { + Some(g) => g, + None => return Frame::Error(Bytes::from_static(b"ERR graph not found")), + }; + + let memgraph = &graph.write_buf; + let lsn = u64::MAX - 1; + + if mode.eq_ignore_ascii_case(b"FILTER") { + // GRAPH.HYBRID g FILTER + if args.len() < 6 { + return Frame::Error(Bytes::from_static( + b"ERR FILTER requires: start_id hops k vector", + )); + } + let start_id = match parse_u64(&args[2]) { + Some(id) => id, + None => return Frame::Error(Bytes::from_static(b"ERR invalid start node ID")), + }; + let hops = match parse_u32(&args[3]) { + Some(h) if h > 0 && h <= 10 => h, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid hops")), + }; + let k = match parse_u32(&args[4]) { + Some(k) if k > 0 => k as usize, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid k")), + }; + let query_vector = match extract_f32_vector(&args[5]) { + Some(v) if !v.is_empty() => v, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid vector")), + }; + + let node_key = super::graph_write::external_id_to_node_key(start_id); + let search = + crate::graph::hybrid::GraphFilteredSearch::new(node_key, hops, query_vector, k); + match search.execute(memgraph, lsn) { + Ok(results) => hybrid_results_to_frame(&results), + Err(e) => Frame::Error(Bytes::from(format!("ERR {e}"))), + } + } else if mode.eq_ignore_ascii_case(b"WALK") { + // GRAPH.HYBRID g WALK + if args.len() < 7 { + return Frame::Error(Bytes::from_static( + b"ERR WALK requires: start_id max_depth beam_width min_sim vector", + )); + } + let start_id = match parse_u64(&args[2]) { + Some(id) => id, + None => return Frame::Error(Bytes::from_static(b"ERR invalid start node ID")), + }; + let max_depth = match parse_u32(&args[3]) { + Some(d) if d > 0 && d <= 100 => d, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid max_depth")), + }; + let beam_width = match parse_u32(&args[4]) { + Some(bw) if bw > 0 => bw as usize, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid beam_width")), + }; + let min_sim = match parse_f64(&args[5]) { + Some(s) => s, + None => return Frame::Error(Bytes::from_static(b"ERR invalid min_sim")), + }; + let query_vector = match extract_f32_vector(&args[6]) { + Some(v) if !v.is_empty() => v, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid vector")), + }; + + let node_key = super::graph_write::external_id_to_node_key(start_id); + let mut walk = + crate::graph::hybrid::VectorGuidedWalk::new(node_key, query_vector, max_depth); + walk.beam_width = beam_width; + walk.min_similarity = min_sim; + + match walk.execute(memgraph, lsn) { + Ok(results) => hybrid_results_to_frame(&results), + Err(e) => Frame::Error(Bytes::from(format!("ERR {e}"))), + } + } else if mode.eq_ignore_ascii_case(b"RERANK") { + // GRAPH.HYBRID g RERANK + if args.len() < 7 { + return Frame::Error(Bytes::from_static( + b"ERR RERANK requires: ref_node_id max_hops alpha k vector", + )); + } + let ref_id = match parse_u64(&args[2]) { + Some(id) => id, + None => return Frame::Error(Bytes::from_static(b"ERR invalid ref node ID")), + }; + let max_hops = match parse_u32(&args[3]) { + Some(h) if h > 0 && h <= 10 => h, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid max_hops (1..10)")), + }; + let alpha = match parse_f64(&args[4]) { + Some(a) => a.clamp(0.0, 1.0), + None => return Frame::Error(Bytes::from_static(b"ERR invalid alpha")), + }; + let k = match parse_u32(&args[5]) { + Some(k) if k > 0 => k as usize, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid k")), + }; + let query_vector = match extract_f32_vector(&args[6]) { + Some(v) if !v.is_empty() => v, + _ => return Frame::Error(Bytes::from_static(b"ERR invalid vector")), + }; + + let node_key = super::graph_write::external_id_to_node_key(ref_id); + let reranker = crate::graph::hybrid::GraphConstrainedReRanker::new( + node_key, + max_hops, + alpha, + query_vector, + k, + ); + match reranker.execute(memgraph, lsn) { + Ok(results) => hybrid_results_to_frame(&results), + Err(e) => Frame::Error(Bytes::from(format!("ERR {e}"))), + } + } else { + Frame::Error(Bytes::from_static( + b"ERR unknown GRAPH.HYBRID mode (supported: FILTER, WALK, RERANK)", + )) + } +} + +// --------------------------------------------------------------------------- +// Hybrid result formatting +// --------------------------------------------------------------------------- + +/// Convert hybrid results to a RESP3 Array of Maps. +fn hybrid_results_to_frame(results: &[crate::graph::hybrid::HybridResult]) -> Frame { + let frames: Vec = results + .iter() + .map(|r| { + let ext_id = r.node.data().as_ffi(); + let mut pairs = vec![ + ( + Frame::SimpleString(Bytes::from_static(b"id")), + Frame::Integer(ext_id as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"score")), + Frame::Double(r.score), + ), + ]; + + if let Some(dist) = r.graph_distance { + pairs.push(( + Frame::SimpleString(Bytes::from_static(b"graph_distance")), + Frame::Integer(dist as i64), + )); + } + + if !r.context.is_empty() { + let ctx: Vec = r + .context + .iter() + .map(|c| { + let ctx_id = c.node.data().as_ffi(); + Frame::Map(vec![ + ( + Frame::SimpleString(Bytes::from_static(b"id")), + Frame::Integer(ctx_id as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"edge_type")), + Frame::Integer(c.edge_type as i64), + ), + ( + Frame::SimpleString(Bytes::from_static(b"hops")), + Frame::Integer(c.hops as i64), + ), + ]) + }) + .collect(); + pairs.push(( + Frame::SimpleString(Bytes::from_static(b"context")), + Frame::Array(ctx.into()), + )); + } + + Frame::Map(pairs) + }) + .collect(); + + Frame::Array(frames.into()) +} + +/// Extract a float vector from a Frame (space-separated f32 values in a BulkString). +fn extract_f32_vector(frame: &Frame) -> Option> { + let bytes = match frame { + Frame::BulkString(b) => b.as_ref(), + Frame::SimpleString(b) => b.as_ref(), + _ => return None, + }; + + let text = core::str::from_utf8(bytes).ok()?; + let values: Result, _> = text.split_whitespace().map(|s| s.parse::()).collect(); + values.ok() +} + +/// Parse an f64 from a Frame. +fn parse_f64(frame: &Frame) -> Option { + match frame { + Frame::Double(f) => Some(*f), + Frame::Integer(n) => Some(*n as f64), + Frame::BulkString(b) | Frame::SimpleString(b) => core::str::from_utf8(b).ok()?.parse().ok(), + _ => None, + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn parse_u64(frame: &Frame) -> Option { + match frame { + Frame::Integer(n) => { + if *n >= 0 { + Some(*n as u64) + } else { + None + } + } + Frame::BulkString(b) | Frame::SimpleString(b) => core::str::from_utf8(b).ok()?.parse().ok(), + _ => None, + } +} + +fn parse_u32(frame: &Frame) -> Option { + match frame { + Frame::Integer(n) => { + if *n >= 0 && *n <= u32::MAX as i64 { + Some(*n as u32) + } else { + None + } + } + Frame::BulkString(b) | Frame::SimpleString(b) => core::str::from_utf8(b).ok()?.parse().ok(), + _ => None, + } +} diff --git a/src/command/graph/graph_write.rs b/src/command/graph/graph_write.rs new file mode 100644 index 00000000..90144eca --- /dev/null +++ b/src/command/graph/graph_write.rs @@ -0,0 +1,484 @@ +//! GRAPH.* write command handlers. +//! +//! These commands mutate GraphStore state: CREATE, ADDNODE, ADDEDGE, DELETE. +//! All operate on the shard-local `NamedGraph.write_buf` (MemGraph) directly, +//! avoiding ArcSwap overhead on the write path. + +use bytes::Bytes; +use slotmap::Key; +use smallvec::SmallVec; + +use crate::graph::store::GraphStore; +use crate::graph::types::{PropertyMap, PropertyValue}; +use crate::graph::wal; +use crate::protocol::Frame; + +/// GRAPH.CREATE +/// +/// Creates a new named graph. Returns OK on success. +/// Error if graph already exists. +pub fn graph_create(store: &mut GraphStore, args: &[Frame]) -> Frame { + if args.is_empty() { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'GRAPH.CREATE' command", + )); + } + + let name = match extract_bulk(&args[0]) { + Some(b) => b, + None => return Frame::Error(Bytes::from_static(b"ERR invalid graph name")), + }; + + // Default edge threshold: 64K edges before freeze. + let edge_threshold = 64_000; + let lsn = store.allocate_lsn(); + + match store.create_graph(Bytes::copy_from_slice(name), edge_threshold, lsn) { + Ok(()) => { + store.wal_pending.push(wal::serialize_graph_create(name)); + Frame::SimpleString(Bytes::from_static(b"OK")) + } + Err(crate::graph::store::GraphStoreError::GraphAlreadyExists) => { + Frame::Error(Bytes::from_static(b"ERR graph already exists")) + } + Err(_) => Frame::Error(Bytes::from_static(b"ERR internal error")), + } +} + +/// GRAPH.ADDNODE