Retire legacy Go organizational graph neighborhood reads

.github/workflows/ephemeral-cerebro.yml

@@ -17,7 +17,7 @@ on:
           - source
           - published
       soak_seconds:
-        description: Sustained shadow-read exercise duration (60-1800 seconds)
+        description: Sustained Rust-authority read exercise duration (60-1800 seconds)
         required: false
         default: 60
         type: number

apps/web/scripts/local-grc-e2e.mjs

@@ -44,8 +44,8 @@ let workDir;
 let logDir;
 let failed = false;
 let backendProcess = null;
+let rustGraphProcess = null;
 let webProcess = null;
-let neo4jStopped = false;
 let postgresStarted = false;
 let neo4jStarted = false;
 let overallDeadlineAt = 0;
@@ -250,38 +250,71 @@ async function main(options) {
   const binarySuffix = process.platform === "win32" ? ".exe" : "";
   const apiBinary = path.join(workDir, `cerebro${binarySuffix}`);
   const eventAdmissionBinary = path.join(backendRoot, "target", "release", `cerebro-event-admission-worker${binarySuffix}`);
+  const rustGraphBinary = path.join(backendRoot, "target", "release", `cerebro-platform${binarySuffix}`);
+  const rustGraphSharedSecret = `grc-e2e-${randomBytes(32).toString("base64url")}`;
   const apiBuild = Promise.all([
     run("go", ["-C", backendRoot, "build", "-o", apiBinary, "./cmd/cerebro"], {
       env: portableChildEnvironment(),
       quiet: true,
     }),
-    run("cargo", ["build", "--locked", "--release", "-p", "cerebro-sourceruntime-eventadmission", "--bin", "cerebro-event-admission-worker"], {
+    run("cargo", [
+      "build", "--locked", "--release",
+      "-p", "cerebro-sourceruntime-eventadmission",
+      "-p", "cerebro-platform",
+      "--bins",
+    ], {
       cwd: backendRoot,
       env: portableChildEnvironment(),
       quiet: true,
     }),
   ]);
-  backendProcess = await startAfterPrerequisite(apiBuild, async () => {
-    const reservation = await reserveDistinctLoopbackPort(usedPorts, { control: activeRunControl });
-    apiPort = reservation.port;
-    usedPorts.add(apiPort);
-    apiBase = `http://127.0.0.1:${apiPort}`;
-    return handoffLoopbackReservation(reservation, () => spawnLogged("cerebro-api", apiBinary, ["serve"], {
+  await apiBuild;
+  const rustGraphReservation = await reserveDistinctLoopbackPort(usedPorts, { control: activeRunControl });
+  const rustGraphPort = rustGraphReservation.port;
+  usedPorts.add(rustGraphPort);
+  const rustGraphBase = `http://127.0.0.1:${rustGraphPort}`;
+  rustGraphProcess = await handoffLoopbackReservation(rustGraphReservation, () => spawnLogged(
+    "cerebro-rust-graph",
+    rustGraphBinary,
+    ["serve-neo4j-readonly"],
+    {
+      cwd: backendRoot,
       env: portableChildEnvironment(process.env, {
-        CEREBRO_HTTP_ADDR: `127.0.0.1:${apiPort}`,
-        CEREBRO_STATE_STORE_DRIVER: "postgres",
-        CEREBRO_POSTGRES_DSN: postgresDSN,
-        CEREBRO_GRAPH_STORE_DRIVER: "neo4j",
         CEREBRO_NEO4J_URI: neo4jURI,
         CEREBRO_NEO4J_USERNAME: neo4jUser,
         CEREBRO_NEO4J_PASSWORD: neo4jCredential,
-        CEREBRO_EVENT_ADMISSION_WORKER: eventAdmissionBinary,
-        CEREBRO_API_AUTH_ENABLED: "false",
-        CEREBRO_DEV_MODE: "1",
-        CEREBRO_DEV_MODE_ACK: "1",
+        CEREBRO_ORGANIZATIONAL_GRAPH_SHARED_SECRET: rustGraphSharedSecret,
+        CEREBRO_RUST_BIND: `127.0.0.1:${rustGraphPort}`,
       }),
-    }), activeRunControl, "Cerebro API spawn");
-  }, activeRunControl, "Cerebro API port reservation");
+    },
+  ), activeRunControl, "Rust graph spawn");
+  await waitFor("Rust graph readiness", async () => {
+    const readiness = await request(`${rustGraphBase}/readyz`);
+    expect(readiness.status === 200, `Rust graph readiness status ${readiness.status}`);
+  }, Math.min(120_000, remainingValidationMs()), rustGraphProcess);
+
+  const apiReservation = await reserveDistinctLoopbackPort(usedPorts, { control: activeRunControl });
+  apiPort = apiReservation.port;
+  usedPorts.add(apiPort);
+  apiBase = `http://127.0.0.1:${apiPort}`;
+  backendProcess = await handoffLoopbackReservation(apiReservation, () => spawnLogged("cerebro-api", apiBinary, ["serve"], {
+    env: portableChildEnvironment(process.env, {
+      CEREBRO_HTTP_ADDR: `127.0.0.1:${apiPort}`,
+      CEREBRO_STATE_STORE_DRIVER: "postgres",
+      CEREBRO_POSTGRES_DSN: postgresDSN,
+      CEREBRO_GRAPH_STORE_DRIVER: "neo4j",
+      CEREBRO_NEO4J_URI: neo4jURI,
+      CEREBRO_NEO4J_USERNAME: neo4jUser,
+      CEREBRO_NEO4J_PASSWORD: neo4jCredential,
+      CEREBRO_ORGANIZATIONAL_GRAPH_READ_URL: rustGraphBase,
+      CEREBRO_ORGANIZATIONAL_GRAPH_READ_MODE: "authority",
+      CEREBRO_ORGANIZATIONAL_GRAPH_SHARED_SECRET: rustGraphSharedSecret,
+      CEREBRO_EVENT_ADMISSION_WORKER: eventAdmissionBinary,
+      CEREBRO_API_AUTH_ENABLED: "false",
+      CEREBRO_DEV_MODE: "1",
+      CEREBRO_DEV_MODE_ACK: "1",
+    }),
+  }), activeRunControl, "Cerebro API spawn");
   await waitFor("Cerebro API readiness", async () => {
     const health = await requestJSON(`${apiBase}/health`);
     expect(health.status === 200, `health status ${health.status}`);
@@ -973,7 +1006,8 @@ async function validateFailureModes() {
   const warmDashboard = await requestJSON(staleDashboardUrl, proxyCacheProbeOptions);
   expect(warmDashboard.status === 200, `warm dashboard status ${warmDashboard.status}`);
   await abortableSleep(proxyCacheTtlMs + 500, activeRunControl?.signal, "proxy cache expiry");
-  await stopNeo4j();
+  await stopChild(rustGraphProcess);
+  rustGraphProcess = null;
   const staleImpact = await requestJSON(staleImpactUrl, proxyCacheProbeOptions);
   expect(staleImpact.status === 200, `stale impact status ${staleImpact.status}`);
   expect(staleImpact.headers.get("x-cerebro-cache") === "stale", `stale impact cache ${staleImpact.headers.get("x-cerebro-cache")}`);
@@ -991,16 +1025,6 @@ async function validateFailureModes() {
   expect(nonCacheableDown.status === 502, `non-cacheable backend-down status ${nonCacheableDown.status}`);
 }
 
-async function stopNeo4j() {
-  if (neo4jStopped || !neo4jStarted) return;
-  neo4jStopped = true;
-  await run("docker", ["stop", "--time", "5", neo4jContainer], {
-    deadlineAt: Math.min(overallDeadlineAt, Date.now() + 10_000),
-    quiet: true,
-  });
-  neo4jStarted = false;
-}
-
 async function stopChild(child) {
   if (!child) return;
   if (child.stopPromise) return child.stopPromise;
@@ -1134,6 +1158,11 @@ async function cleanup() {
   } catch (error) {
     errors.push(error);
   }
+  try {
+    await stopChild(rustGraphProcess);
+  } catch (error) {
+    errors.push(error);
+  }
   if (neo4jStarted) {
     try {
       await run("docker", ["rm", "-f", neo4jContainer], { deadlineAt: overallDeadlineAt, quiet: true, signal: null });
@@ -1203,13 +1232,13 @@ export async function runLocalGrcE2E(options = {}) {
   activeRunControl = createRunControl(validationDeadlineAt);
   failed = false;
   backendProcess = null;
+  rustGraphProcess = null;
   webProcess = null;
   delayRelay = null;
   workDir = undefined;
   logDir = undefined;
   apiBase = undefined;
   webBase = undefined;
-  neo4jStopped = false;
   postgresStarted = false;
   neo4jStarted = false;
 

apps/web/scripts/local-grc-e2e.test.mjs

@@ -126,6 +126,8 @@ describe("real-service E2E isolation", () => {
     const seed = await readFile(seedPath, "utf8");
     expect(seed).toContain("e2e-local");
     expect(seed).toContain("example.org");
+    expect(seed).toContain(":OrganizationalEntity");
+    expect(seed).toContain(":ORGANIZATIONAL_RELATION");
     expect(seed).not.toMatch(/\breplace\s|GOPRIVATE|example\.com|unused-local/);
   });
 
@@ -144,6 +146,15 @@ describe("real-service E2E isolation", () => {
     expect(runner).toContain("CEREBRO_EVENT_ADMISSION_WORKER: eventAdmissionBinary");
   });
 
+  it("routes product graph reads through the Rust authority", async () => {
+    const runnerPath = fileURLToPath(new URL("./local-grc-e2e.mjs", import.meta.url));
+    const runner = await readFile(runnerPath, "utf8");
+    expect(runner).toContain('"serve-neo4j-readonly"');
+    expect(runner).toContain("CEREBRO_ORGANIZATIONAL_GRAPH_READ_URL: rustGraphBase");
+    expect(runner).toContain('CEREBRO_ORGANIZATIONAL_GRAPH_READ_MODE: "authority"');
+    expect(runner).toContain("CEREBRO_ORGANIZATIONAL_GRAPH_SHARED_SECRET: rustGraphSharedSecret");
+  });
+
   it("requires visible seeded data and a successful API response in Chromium", () => {
     expect(browserDataContract("/risk-inbox")).toEqual({
       apiPath: "/api/cerebro/grc/findings",

apps/web/scripts/testdata/grc-e2e-seed/main.go

@@ -2,10 +2,12 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"time"
 
+	neo4jdriver "github.com/neo4j/neo4j-go-driver/v5/neo4j"
 	cerebrov1 "github.com/writer/cerebro/gen/cerebro/v1"
 	"github.com/writer/cerebro/internal/config"
 	findinganalysis "github.com/writer/cerebro/internal/findings"
@@ -88,9 +90,111 @@ func seedNeo4j(ctx context.Context) {
 	for _, link := range links {
 		must(store.UpsertProjectedLink(ctx, link))
 	}
-	neighborhood, err := store.GetEntityNeighborhood(ctx, adminURN, 10)
+	seedRustAuthorityGraph(ctx, entities, links)
+	rows, err := store.ExecuteReadCypher(ctx, ports.CypherQueryRequest{
+		Query: `MATCH (root:Entity {urn: $root_urn})-[relation:RELATION]-(neighbor:Entity)
+RETURN count(DISTINCT neighbor) AS neighbors, count(relation) AS relations`,
+		Params:   map[string]any{"root_urn": adminURN},
+		RowLimit: 1,
+	})
 	must(err)
-	if neighborhood.Root == nil || len(neighborhood.Neighbors) < 2 || len(neighborhood.Relations) < 2 {
-		panic("seeded graph neighborhood is incomplete")
+	if len(rows) != 1 || fmt.Sprint(rows[0].Values["neighbors"]) == "0" || fmt.Sprint(rows[0].Values["relations"]) == "0" {
+		panic("seeded graph relations are incomplete")
 	}
 }
+
+func seedRustAuthorityGraph(ctx context.Context, entities []*ports.ProjectedEntity, links []*ports.ProjectedLink) {
+	driver, err := neo4jdriver.NewDriverWithContext(
+		os.Getenv("CEREBRO_NEO4J_URI"),
+		neo4jdriver.BasicAuth(
+			os.Getenv("CEREBRO_NEO4J_USERNAME"),
+			os.Getenv("CEREBRO_NEO4J_PASSWORD"),
+			"",
+		),
+	)
+	must(err)
+	defer func() { _ = driver.Close(ctx) }()
+
+	entityIDs := map[string]string{
+		adminURN: "identity-privileged",
+		appURN:   "application-admin-console",
+		repoURN:  "repository-public-protected",
+		guestURN: "identity-external-collaborator",
+	}
+	entityRows := make([]map[string]any, 0, len(entities))
+	for _, entity := range entities {
+		properties, err := json.Marshal(map[string]string{"entity_urn": entity.URN})
+		must(err)
+		entityRows = append(entityRows, map[string]any{
+			"authority_json":  "{}",
+			"entity_id":       entityIDs[entity.URN],
+			"entity_kind":     entity.EntityType,
+			"external_id":     entity.URN,
+			"label":           entity.Label,
+			"properties_json": string(properties),
+		})
+	}
+	linkRows := make([]map[string]any, 0, len(links))
+	for index, link := range links {
+		linkRows = append(linkRows, map[string]any{
+			"assertion_id":      fmt.Sprintf("grc-e2e-%d", index+1),
+			"from_entity_id":    entityIDs[link.FromURN],
+			"relation":          link.Relation,
+			"source_runtime_id": link.RuntimeID,
+			"to_entity_id":      entityIDs[link.ToURN],
+		})
+	}
+
+	session := driver.NewSession(ctx, neo4jdriver.SessionConfig{})
+	defer func() { _ = session.Close(ctx) }()
+	_, err = session.ExecuteWrite(ctx, func(tx neo4jdriver.ManagedTransaction) (any, error) {
+		result, err := tx.Run(ctx, `UNWIND $entities AS row
+MATCH (entity:Entity {urn: row.external_id})
+SET entity:OrganizationalEntity,
+    entity.tenant_id = $tenant_id,
+    entity.entity_id = row.entity_id,
+    entity.entity_kind = row.entity_kind,
+    entity.authority_json = row.authority_json,
+    entity.label = row.label,
+    entity.properties_json = row.properties_json,
+    entity.external_id = row.external_id
+WITH count(entity) AS projected
+MERGE (revision:OrganizationalGraphRevision {tenant_id: $tenant_id})
+SET revision.graph_revision = 1
+RETURN projected`, map[string]any{"entities": entityRows, "tenant_id": tenantID})
+		if err != nil {
+			return nil, err
+		}
+		record, err := result.Single(ctx)
+		if err != nil {
+			return nil, err
+		}
+		projected, ok := record.Values[0].(int64)
+		if !ok || projected != int64(len(entityRows)) {
+			return nil, fmt.Errorf("projected Rust authority entities = %v, want %d", record.Values[0], len(entityRows))
+		}
+		result, err = tx.Run(ctx, `UNWIND $links AS row
+MATCH (source:OrganizationalEntity {tenant_id: $tenant_id, entity_id: row.from_entity_id})
+MATCH (target:OrganizationalEntity {tenant_id: $tenant_id, entity_id: row.to_entity_id})
+MERGE (source)-[relation:ORGANIZATIONAL_RELATION {
+  tenant_id: $tenant_id,
+  assertion_id: row.assertion_id
+}]->(target)
+SET relation.relation = row.relation,
+    relation.source_runtime_id = row.source_runtime_id
+RETURN count(relation) AS projected`, map[string]any{"links": linkRows, "tenant_id": tenantID})
+		if err != nil {
+			return nil, err
+		}
+		record, err = result.Single(ctx)
+		if err != nil {
+			return nil, err
+		}
+		projected, ok = record.Values[0].(int64)
+		if !ok || projected != int64(len(linkRows)) {
+			return nil, fmt.Errorf("projected Rust authority relations = %v, want %d", record.Values[0], len(linkRows))
+		}
+		return nil, nil
+	})
+	must(err)
+}

cmd/cerebro/aws_github_neo4j_live_e2e_test.go

@@ -85,7 +85,7 @@ func TestAWSGitHubNeo4jSharedIdentityLiveE2E(t *testing.T) {
 		t.Fatalf("graph counts = %#v, want non-zero nodes and relations", counts)
 	}
 	identityURN := "urn:cerebro:" + tenantID + ":identity:email:" + strings.ToLower(sharedEmail)
-	neighborhood, err := store.GetEntityNeighborhood(ctx, identityURN, 50)
+	neighborhood, err := readNeo4jLiveNeighborhood(ctx, store, identityURN, 50)
 	if err != nil {
 		t.Fatalf("GetEntityNeighborhood(%q) error = %v", identityURN, err)
 	}

cmd/cerebro/github_findings_live_e2e_test.go

@@ -14,7 +14,6 @@ import (
 
 	cerebrov1 "github.com/writer/cerebro/gen/cerebro/v1"
 	"github.com/writer/cerebro/internal/findings"
-	"github.com/writer/cerebro/internal/graphquery"
 	"github.com/writer/cerebro/internal/ports"
 	"github.com/writer/cerebro/internal/sourceops"
 	"github.com/writer/cerebro/internal/sourceprojection"
@@ -119,10 +118,7 @@ func TestGitHubDependabotFindingsEndToEndWithGHCLI(t *testing.T) {
 		t.Fatalf("ResolveFinding().Status = %q, want resolved", resolved.Status)
 	}
 
-	alertNeighborhood, err := graphquery.New(graphStore).GetEntityNeighborhood(ctx, graphquery.NeighborhoodRequest{
-		RootURN: primaryResourceURN,
-		Limit:   20,
-	})
+	alertNeighborhood, err := readNeo4jLiveNeighborhood(ctx, graphStore, primaryResourceURN, 20)
 	if err != nil {
 		t.Fatalf("GetEntityNeighborhood(%q) error = %v", primaryResourceURN, err)
 	}
@@ -136,10 +132,7 @@ func TestGitHubDependabotFindingsEndToEndWithGHCLI(t *testing.T) {
 	}
 
 	findingURN := "urn:cerebro:" + finding.TenantID + ":finding:" + finding.ID
-	findingNeighborhood, err := graphquery.New(graphStore).GetEntityNeighborhood(ctx, graphquery.NeighborhoodRequest{
-		RootURN: findingURN,
-		Limit:   20,
-	})
+	findingNeighborhood, err := readNeo4jLiveNeighborhood(ctx, graphStore, findingURN, 20)
 	if err != nil {
 		t.Fatalf("GetEntityNeighborhood(%q) error = %v", findingURN, err)
 	}
@@ -260,10 +253,7 @@ func TestGitHubAuditFindingsGraphPreviewWithGHCLI(t *testing.T) {
 	neighborhoods := make([]graphPreviewNeighborhood, 0, len(previewFindings))
 	neighborhoodsByFindingURN := map[string]*ports.EntityNeighborhood{}
 	for _, finding := range allPreviewFindings {
-		neighborhood, err := graphquery.New(graphStore).GetEntityNeighborhood(ctx, graphquery.NeighborhoodRequest{
-			RootURN: finding.FindingURN,
-			Limit:   50,
-		})
+		neighborhood, err := readNeo4jLiveNeighborhood(ctx, graphStore, finding.FindingURN, 50)
 		if err != nil {
 			t.Fatalf("GetEntityNeighborhood(%q) error = %v", finding.FindingURN, err)
 		}