From 1d7eb33169e36fdb37cb84cf6c920fe5b42f6539 Mon Sep 17 00:00:00 2001
From: John Gemignani <jrgemignani@gmail.com>
Date: Thu, 3 Sep 2026 17:00:08 -0700
Subject: [PATCH] Add RLS and SELECT privilege enforcement to global graph

The global graph cache that backs variable-length edge (VLE) and shortest-path
traversal is populated with a direct heap scan (table_beginscan + heap_getnext).
That scan does not go through the planner or executor, so the row-level security
policies and table privileges that MATCH receives automatically are not applied
to traversal: it reads every vertex and edge regardless of the current role's
policies or grants.

Add age.enforce_rls_in_traversal (bool, PGC_SUSET, default on). When on:

* SELECT privilege is checked on every label table before it is read
  (pg_class_aclcheck, with a pg_attribute_aclcheck fallback so column-level
  grants are honored the same way the executor honors them).

* When RLS is active on a label for the current role, that label is loaded
  through SPI (SELECT ... FROM ONLY <label>) with a batched read-only cursor so
  the planner applies the policies. Otherwise the direct-scan path is used, now
  behind the privilege check. Properties are still fetched lazily by TID: only
  policy-visible rows are loaded, so the later heap_fetch reads only authorized
  tuples, which covers shortest_path as well.

When off, the previous direct-scan behavior is restored, leaving operators a
single escape hatch.

To match MATCH's inner-join behavior, the edge loaders drop any edge whose start
or end vertex was filtered out by a vertex policy rather than leaving a dangling
edge in the cache.

Cache validity:

* A cache that applied RLS is not reused across statements (loaded_with_rls),
  because policies can depend on session state such as current_setting() that is
  not covered by the version, snapshot, role, and GUC cache keys. Graphs that do
  not use RLS are unaffected.

* is_ggctx_invalid checks the role and GUC keys before the graph version-counter
  fast path, so a role or GUC change forces a rebuild even in DSM/SHMEM mode.

* The version counter tracks only graph data changes, so an enforced non-RLS
  cache could otherwise survive a GRANT/REVOKE or policy/RLS DDL on a label table
  and serve stale permissions. A relcache callback keyed by label-table OID
  (table/column privileges, policies, RLS flags), plus pg_authid and
  pg_auth_members syscache callbacks (BYPASSRLS, role membership), now mark such
  caches stale so they are rebuilt and re-authorized.

Re-entrancy and cleanup:

* The context is built detached and attached only after a successful load, so a
  policy expression that re-enters traversal cannot observe a half-built cache,
  and a failed load frees the partial context instead of leaving it in
  TopMemoryContext.

* On attach, an existing context for the same graph is reused only if it is still
  valid for the current role and GUC, so a re-entrant traversal reached through a
  SECURITY DEFINER policy cannot be handed a context loaded as another role. If
  discarding an incompatible context reports a missing entry, that error is
  raised after first freeing the freshly built context.

* Each loader frees the transient label-name list from get_ag_labels_names() on
  both the normal and error paths; it is allocated in TopMemoryContext and would
  otherwise accumulate on every rebuild.

Adds regression test rls_vle covering policy and privilege enforcement, the
opt-out GUC, per-role cache invalidation, a session current_setting() policy,
and privilege-metadata invalidation.

Co-authored-by: GitHub Copilot (Claude Opus 5) <noreply@githubcopilot.com>
---
 Makefile                                 |   1 +
 regress/expected/rls_vle.out             | 369 +++++++++
 regress/sql/rls_vle.sql                  | 264 +++++++
 src/backend/utils/adt/age_global_graph.c | 935 +++++++++++++++++++----
 src/backend/utils/ag_guc.c               |  11 +
 src/include/utils/ag_guc.h               |   9 +
 6 files changed, 1436 insertions(+), 153 deletions(-)
 create mode 100644 regress/expected/rls_vle.out
 create mode 100644 regress/sql/rls_vle.sql

diff --git a/Makefile b/Makefile
index 2ea8f64ed..fac0511c7 100644
--- a/Makefile
+++ b/Makefile
@@ -225,6 +225,7 @@ REGRESS = scan \
           direct_field_access \
           generated_columns \
           security \
+          rls_vle \
           reserved_keyword_alias \
           agtype_jsonb_cast \
           containment_selectivity \
diff --git a/regress/expected/rls_vle.out b/regress/expected/rls_vle.out
new file mode 100644
index 000000000..a3f16caed
--- /dev/null
+++ b/regress/expected/rls_vle.out
@@ -0,0 +1,369 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+LOAD 'age';
+SET search_path TO ag_catalog;
+--
+-- Row-level security enforcement during VLE / global-graph traversal
+-- (controlled by age.enforce_rls_in_traversal, default on).
+--
+-- Chain: n1 -[LINK visible:true]-> n2 -[LINK visible:false]-> n3 -[LINK visible:true]-> n4
+--
+SELECT create_graph('rls_vle');
+NOTICE:  graph "rls_vle" has been created
+ create_graph 
+--------------
+ 
+(1 row)
+
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n1'}) $$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n2'}) $$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n3'}) $$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n4'}) $$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'}), (b:Node {name: 'n2'})
+    CREATE (a)-[:LINK {visible: true}]->(b)
+$$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n2'}), (b:Node {name: 'n3'})
+    CREATE (a)-[:LINK {visible: false}]->(b)
+$$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n3'}), (b:Node {name: 'n4'})
+    CREATE (a)-[:LINK {visible: true}]->(b)
+$$) AS (a agtype);
+ a 
+---
+(0 rows)
+
+-- restricted role: SELECT on the graph + ag_catalog, nothing more
+CREATE ROLE rls_vle_user LOGIN;
+GRANT USAGE ON SCHEMA rls_vle TO rls_vle_user;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_vle TO rls_vle_user;
+GRANT USAGE ON SCHEMA ag_catalog TO rls_vle_user;
+GRANT SELECT ON ALL TABLES IN SCHEMA ag_catalog TO rls_vle_user;
+-- ============================================================================
+-- PART 1: Baseline (superuser, RLS bypassed) -- VLE from n1 reaches n2, n3, n4.
+-- ============================================================================
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+ "n3"
+ "n4"
+(3 rows)
+
+-- ============================================================================
+-- PART 2: Edge-label RLS -- hide edges whose "visible" property is false.
+-- ============================================================================
+ALTER TABLE rls_vle."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" FORCE ROW LEVEL SECURITY;
+CREATE POLICY link_visible ON rls_vle."LINK"
+    FOR SELECT USING ((properties->>'"visible"')::boolean = true);
+-- Enforced (default): the restricted role reaches ONLY n2 (n2->n3 hidden).
+SET ROLE rls_vle_user;
+SHOW age.enforce_rls_in_traversal;
+ age.enforce_rls_in_traversal 
+------------------------------
+ on
+(1 row)
+
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+(1 row)
+
+RESET ROLE;
+-- Opt-out: with enforcement off, the legacy direct scan bypasses RLS, so the
+-- restricted role again reaches n2, n3, n4.
+SET age.enforce_rls_in_traversal = off;
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+ "n3"
+ "n4"
+(3 rows)
+
+RESET ROLE;
+RESET age.enforce_rls_in_traversal;
+-- clear the edge policy for the next part
+DROP POLICY link_visible ON rls_vle."LINK";
+ALTER TABLE rls_vle."LINK" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" DISABLE ROW LEVEL SECURITY;
+-- ============================================================================
+-- PART 3: Vertex-label RLS -- exercises the vertex loader and the dropping of
+-- edges whose endpoint vertex is hidden (matches MATCH's inner-join behavior).
+-- All edges are visible here; only vertices are filtered.
+-- ============================================================================
+ALTER TABLE rls_vle."Node" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."Node" FORCE ROW LEVEL SECURITY;
+-- 3.1: hide the INTERMEDIATE vertex n2. Its edges are dropped, so the traversal
+-- cannot pass through it -- expect no rows (and no "no vertex found" error).
+CREATE POLICY node_visible ON rls_vle."Node"
+    FOR SELECT USING ((properties->>'"name"') <> 'n2');
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+(0 rows)
+
+RESET ROLE;
+-- 3.2: hide only the LEAF vertex n4. The edge into n4 is dropped; the traversal
+-- still reaches n2 and n3.
+DROP POLICY node_visible ON rls_vle."Node";
+CREATE POLICY node_visible ON rls_vle."Node"
+    FOR SELECT USING ((properties->>'"name"') <> 'n4');
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+ "n3"
+(2 rows)
+
+RESET ROLE;
+DROP POLICY node_visible ON rls_vle."Node";
+ALTER TABLE rls_vle."Node" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."Node" DISABLE ROW LEVEL SECURITY;
+-- ============================================================================
+-- PART 4: Per-role cache key -- two roles with different edge policies in one
+-- session must see different results (the cache is rebuilt when the role
+-- changes). user1 sees all edges; user2 sees none.
+-- ============================================================================
+CREATE ROLE rls_vle_user2 LOGIN;
+GRANT USAGE ON SCHEMA rls_vle TO rls_vle_user2;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_vle TO rls_vle_user2;
+GRANT USAGE ON SCHEMA ag_catalog TO rls_vle_user2;
+GRANT SELECT ON ALL TABLES IN SCHEMA ag_catalog TO rls_vle_user2;
+ALTER TABLE rls_vle."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" FORCE ROW LEVEL SECURITY;
+CREATE POLICY link_user1 ON rls_vle."LINK" FOR SELECT TO rls_vle_user  USING (true);
+CREATE POLICY link_user2 ON rls_vle."LINK" FOR SELECT TO rls_vle_user2 USING (false);
+-- user1: all edges visible -> n2, n3, n4
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+ "n3"
+ "n4"
+(3 rows)
+
+RESET ROLE;
+-- user2: no edges visible -> empty (cache must not be reused from user1)
+SET ROLE rls_vle_user2;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+(0 rows)
+
+RESET ROLE;
+-- tidy up Part 4's edge policies before the next parts
+DROP POLICY link_user1 ON rls_vle."LINK";
+DROP POLICY link_user2 ON rls_vle."LINK";
+ALTER TABLE rls_vle."LINK" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" DISABLE ROW LEVEL SECURITY;
+-- ============================================================================
+-- PART 5: SELECT-ACL enforcement -- the low-level loader must not bypass table
+-- privileges, not only RLS policies. rls_vle_user3 can read every label table
+-- EXCEPT the edge table "LINK".
+-- ============================================================================
+CREATE ROLE rls_vle_user3 LOGIN;
+GRANT USAGE ON SCHEMA rls_vle TO rls_vle_user3;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_vle TO rls_vle_user3;
+GRANT USAGE ON SCHEMA ag_catalog TO rls_vle_user3;
+GRANT SELECT ON ALL TABLES IN SCHEMA ag_catalog TO rls_vle_user3;
+REVOKE SELECT ON rls_vle."LINK" FROM rls_vle_user3;
+-- Enforced (default): traversal fails closed with permission denied on "LINK".
+-- The start-node MATCH on "Node" is allowed; only the edge load is denied.
+SET ROLE rls_vle_user3;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ERROR:  permission denied for table LINK
+RESET ROLE;
+-- Opt-out: with enforcement off, the legacy direct scan bypasses the ACL, so
+-- the same role again reaches n2, n3, n4 (the legacy behavior).
+SET age.enforce_rls_in_traversal = off;
+SET ROLE rls_vle_user3;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+ "n3"
+ "n4"
+(3 rows)
+
+RESET ROLE;
+RESET age.enforce_rls_in_traversal;
+-- ============================================================================
+-- PART 6: Session-dependent RLS policy -- a policy whose result depends on a
+-- session GUC (current_setting) must be re-evaluated every statement, so an
+-- RLS-enforced cache is never reused across statements even when the role, the
+-- enforcement GUC, and the snapshot are unchanged.
+-- ============================================================================
+ALTER TABLE rls_vle."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" FORCE ROW LEVEL SECURITY;
+CREATE POLICY link_by_setting ON rls_vle."LINK"
+    FOR SELECT USING (current_setting('rls_vle.mode', true) = 'all');
+SET ROLE rls_vle_user;
+-- mode=all: every edge visible -> n2, n3, n4
+SET rls_vle.mode = 'all';
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "n2"
+ "n3"
+ "n4"
+(3 rows)
+
+-- mode changed in the same session (same role / GUC / snapshot): the cache must
+-- be rebuilt, so now no edge is visible -> empty.
+SET rls_vle.mode = 'none';
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+(0 rows)
+
+RESET rls_vle.mode;
+RESET ROLE;
+DROP POLICY link_by_setting ON rls_vle."LINK";
+ALTER TABLE rls_vle."LINK" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" DISABLE ROW LEVEL SECURITY;
+-- ============================================================================
+-- PART 7: ACL-metadata cache invalidation -- an enforced (non-RLS) direct-scan
+-- cache must not survive a REVOKE on a label table. In DSM/SHMEM mode the graph
+-- version fast path only tracks data changes, so the cache is invalidated via
+-- catalog-invalidation callbacks; without them the REVOKE'd rows stay readable.
+-- Uses a fresh graph with no RLS so the direct-scan (ACL-only) path is taken.
+-- ============================================================================
+SET age.enforce_rls_in_traversal = on;
+SELECT create_graph('rls_acl');
+NOTICE:  graph "rls_acl" has been created
+ create_graph 
+--------------
+ 
+(1 row)
+
+SELECT * FROM cypher('rls_acl', $$
+    CREATE (:Node {name: 'a'})-[:LINK]->(:Node {name: 'b'})-[:LINK]->(:Node {name: 'c'})
+$$) AS (x agtype);
+ x 
+---
+(0 rows)
+
+GRANT USAGE ON SCHEMA rls_acl TO rls_vle_user;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_acl TO rls_vle_user;
+-- build the enforced cache: rls_vle_user has SELECT -> reaches b, c
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_acl', $$
+    MATCH (a:Node {name: 'a'})-[:LINK*1..2]->(x) RETURN x.name
+$$) AS (name agtype) ORDER BY name;
+ name 
+------
+ "b"
+ "c"
+(2 rows)
+
+RESET ROLE;
+-- REVOKE SELECT on the edge label -- no graph data change, version counter same
+REVOKE SELECT ON rls_acl."LINK" FROM rls_vle_user;
+-- the cached traversal must now be rebuilt and denied
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_acl', $$
+    MATCH (a:Node {name: 'a'})-[:LINK*1..2]->(x) RETURN x.name
+$$) AS (name agtype) ORDER BY name;
+ERROR:  permission denied for table LINK
+RESET ROLE;
+SELECT drop_graph('rls_acl', true);
+NOTICE:  drop cascades to 4 other objects
+DETAIL:  drop cascades to table rls_acl._ag_label_vertex
+drop cascades to table rls_acl._ag_label_edge
+drop cascades to table rls_acl."Node"
+drop cascades to table rls_acl."LINK"
+NOTICE:  graph "rls_acl" has been dropped
+ drop_graph 
+------------
+ 
+(1 row)
+
+RESET age.enforce_rls_in_traversal;
+-- ============================================================================
+-- cleanup
+-- ============================================================================
+SELECT drop_graph('rls_vle', true);
+NOTICE:  drop cascades to 4 other objects
+DETAIL:  drop cascades to table rls_vle._ag_label_vertex
+drop cascades to table rls_vle._ag_label_edge
+drop cascades to table rls_vle."Node"
+drop cascades to table rls_vle."LINK"
+NOTICE:  graph "rls_vle" has been dropped
+ drop_graph 
+------------
+ 
+(1 row)
+
+DROP OWNED BY rls_vle_user, rls_vle_user2, rls_vle_user3;
+DROP ROLE rls_vle_user, rls_vle_user2, rls_vle_user3;
diff --git a/regress/sql/rls_vle.sql b/regress/sql/rls_vle.sql
new file mode 100644
index 000000000..d171acd97
--- /dev/null
+++ b/regress/sql/rls_vle.sql
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+LOAD 'age';
+SET search_path TO ag_catalog;
+
+--
+-- Row-level security enforcement during VLE / global-graph traversal
+-- (controlled by age.enforce_rls_in_traversal, default on).
+--
+-- Chain: n1 -[LINK visible:true]-> n2 -[LINK visible:false]-> n3 -[LINK visible:true]-> n4
+--
+
+SELECT create_graph('rls_vle');
+
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n1'}) $$) AS (a agtype);
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n2'}) $$) AS (a agtype);
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n3'}) $$) AS (a agtype);
+SELECT * FROM cypher('rls_vle', $$ CREATE (:Node {name: 'n4'}) $$) AS (a agtype);
+
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'}), (b:Node {name: 'n2'})
+    CREATE (a)-[:LINK {visible: true}]->(b)
+$$) AS (a agtype);
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n2'}), (b:Node {name: 'n3'})
+    CREATE (a)-[:LINK {visible: false}]->(b)
+$$) AS (a agtype);
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n3'}), (b:Node {name: 'n4'})
+    CREATE (a)-[:LINK {visible: true}]->(b)
+$$) AS (a agtype);
+
+-- restricted role: SELECT on the graph + ag_catalog, nothing more
+CREATE ROLE rls_vle_user LOGIN;
+GRANT USAGE ON SCHEMA rls_vle TO rls_vle_user;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_vle TO rls_vle_user;
+GRANT USAGE ON SCHEMA ag_catalog TO rls_vle_user;
+GRANT SELECT ON ALL TABLES IN SCHEMA ag_catalog TO rls_vle_user;
+
+-- ============================================================================
+-- PART 1: Baseline (superuser, RLS bypassed) -- VLE from n1 reaches n2, n3, n4.
+-- ============================================================================
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+
+-- ============================================================================
+-- PART 2: Edge-label RLS -- hide edges whose "visible" property is false.
+-- ============================================================================
+ALTER TABLE rls_vle."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" FORCE ROW LEVEL SECURITY;
+CREATE POLICY link_visible ON rls_vle."LINK"
+    FOR SELECT USING ((properties->>'"visible"')::boolean = true);
+
+-- Enforced (default): the restricted role reaches ONLY n2 (n2->n3 hidden).
+SET ROLE rls_vle_user;
+SHOW age.enforce_rls_in_traversal;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+-- Opt-out: with enforcement off, the legacy direct scan bypasses RLS, so the
+-- restricted role again reaches n2, n3, n4.
+SET age.enforce_rls_in_traversal = off;
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+RESET age.enforce_rls_in_traversal;
+
+-- clear the edge policy for the next part
+DROP POLICY link_visible ON rls_vle."LINK";
+ALTER TABLE rls_vle."LINK" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" DISABLE ROW LEVEL SECURITY;
+
+-- ============================================================================
+-- PART 3: Vertex-label RLS -- exercises the vertex loader and the dropping of
+-- edges whose endpoint vertex is hidden (matches MATCH's inner-join behavior).
+-- All edges are visible here; only vertices are filtered.
+-- ============================================================================
+ALTER TABLE rls_vle."Node" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."Node" FORCE ROW LEVEL SECURITY;
+
+-- 3.1: hide the INTERMEDIATE vertex n2. Its edges are dropped, so the traversal
+-- cannot pass through it -- expect no rows (and no "no vertex found" error).
+CREATE POLICY node_visible ON rls_vle."Node"
+    FOR SELECT USING ((properties->>'"name"') <> 'n2');
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+-- 3.2: hide only the LEAF vertex n4. The edge into n4 is dropped; the traversal
+-- still reaches n2 and n3.
+DROP POLICY node_visible ON rls_vle."Node";
+CREATE POLICY node_visible ON rls_vle."Node"
+    FOR SELECT USING ((properties->>'"name"') <> 'n4');
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+DROP POLICY node_visible ON rls_vle."Node";
+ALTER TABLE rls_vle."Node" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."Node" DISABLE ROW LEVEL SECURITY;
+
+-- ============================================================================
+-- PART 4: Per-role cache key -- two roles with different edge policies in one
+-- session must see different results (the cache is rebuilt when the role
+-- changes). user1 sees all edges; user2 sees none.
+-- ============================================================================
+CREATE ROLE rls_vle_user2 LOGIN;
+GRANT USAGE ON SCHEMA rls_vle TO rls_vle_user2;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_vle TO rls_vle_user2;
+GRANT USAGE ON SCHEMA ag_catalog TO rls_vle_user2;
+GRANT SELECT ON ALL TABLES IN SCHEMA ag_catalog TO rls_vle_user2;
+
+ALTER TABLE rls_vle."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" FORCE ROW LEVEL SECURITY;
+CREATE POLICY link_user1 ON rls_vle."LINK" FOR SELECT TO rls_vle_user  USING (true);
+CREATE POLICY link_user2 ON rls_vle."LINK" FOR SELECT TO rls_vle_user2 USING (false);
+
+-- user1: all edges visible -> n2, n3, n4
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+-- user2: no edges visible -> empty (cache must not be reused from user1)
+SET ROLE rls_vle_user2;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+-- tidy up Part 4's edge policies before the next parts
+DROP POLICY link_user1 ON rls_vle."LINK";
+DROP POLICY link_user2 ON rls_vle."LINK";
+ALTER TABLE rls_vle."LINK" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" DISABLE ROW LEVEL SECURITY;
+
+-- ============================================================================
+-- PART 5: SELECT-ACL enforcement -- the low-level loader must not bypass table
+-- privileges, not only RLS policies. rls_vle_user3 can read every label table
+-- EXCEPT the edge table "LINK".
+-- ============================================================================
+CREATE ROLE rls_vle_user3 LOGIN;
+GRANT USAGE ON SCHEMA rls_vle TO rls_vle_user3;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_vle TO rls_vle_user3;
+GRANT USAGE ON SCHEMA ag_catalog TO rls_vle_user3;
+GRANT SELECT ON ALL TABLES IN SCHEMA ag_catalog TO rls_vle_user3;
+REVOKE SELECT ON rls_vle."LINK" FROM rls_vle_user3;
+
+-- Enforced (default): traversal fails closed with permission denied on "LINK".
+-- The start-node MATCH on "Node" is allowed; only the edge load is denied.
+SET ROLE rls_vle_user3;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+-- Opt-out: with enforcement off, the legacy direct scan bypasses the ACL, so
+-- the same role again reaches n2, n3, n4 (the legacy behavior).
+SET age.enforce_rls_in_traversal = off;
+SET ROLE rls_vle_user3;
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+RESET age.enforce_rls_in_traversal;
+
+-- ============================================================================
+-- PART 6: Session-dependent RLS policy -- a policy whose result depends on a
+-- session GUC (current_setting) must be re-evaluated every statement, so an
+-- RLS-enforced cache is never reused across statements even when the role, the
+-- enforcement GUC, and the snapshot are unchanged.
+-- ============================================================================
+ALTER TABLE rls_vle."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" FORCE ROW LEVEL SECURITY;
+CREATE POLICY link_by_setting ON rls_vle."LINK"
+    FOR SELECT USING (current_setting('rls_vle.mode', true) = 'all');
+
+SET ROLE rls_vle_user;
+-- mode=all: every edge visible -> n2, n3, n4
+SET rls_vle.mode = 'all';
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+-- mode changed in the same session (same role / GUC / snapshot): the cache must
+-- be rebuilt, so now no edge is visible -> empty.
+SET rls_vle.mode = 'none';
+SELECT * FROM cypher('rls_vle', $$
+    MATCH (a:Node {name: 'n1'})-[:LINK*1..3]->(b) RETURN b.name
+$$) AS (name agtype) ORDER BY name;
+RESET rls_vle.mode;
+RESET ROLE;
+
+DROP POLICY link_by_setting ON rls_vle."LINK";
+ALTER TABLE rls_vle."LINK" NO FORCE ROW LEVEL SECURITY;
+ALTER TABLE rls_vle."LINK" DISABLE ROW LEVEL SECURITY;
+
+-- ============================================================================
+-- PART 7: ACL-metadata cache invalidation -- an enforced (non-RLS) direct-scan
+-- cache must not survive a REVOKE on a label table. In DSM/SHMEM mode the graph
+-- version fast path only tracks data changes, so the cache is invalidated via
+-- catalog-invalidation callbacks; without them the REVOKE'd rows stay readable.
+-- Uses a fresh graph with no RLS so the direct-scan (ACL-only) path is taken.
+-- ============================================================================
+SET age.enforce_rls_in_traversal = on;
+SELECT create_graph('rls_acl');
+SELECT * FROM cypher('rls_acl', $$
+    CREATE (:Node {name: 'a'})-[:LINK]->(:Node {name: 'b'})-[:LINK]->(:Node {name: 'c'})
+$$) AS (x agtype);
+GRANT USAGE ON SCHEMA rls_acl TO rls_vle_user;
+GRANT SELECT ON ALL TABLES IN SCHEMA rls_acl TO rls_vle_user;
+
+-- build the enforced cache: rls_vle_user has SELECT -> reaches b, c
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_acl', $$
+    MATCH (a:Node {name: 'a'})-[:LINK*1..2]->(x) RETURN x.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+-- REVOKE SELECT on the edge label -- no graph data change, version counter same
+REVOKE SELECT ON rls_acl."LINK" FROM rls_vle_user;
+
+-- the cached traversal must now be rebuilt and denied
+SET ROLE rls_vle_user;
+SELECT * FROM cypher('rls_acl', $$
+    MATCH (a:Node {name: 'a'})-[:LINK*1..2]->(x) RETURN x.name
+$$) AS (name agtype) ORDER BY name;
+RESET ROLE;
+
+SELECT drop_graph('rls_acl', true);
+RESET age.enforce_rls_in_traversal;
+
+-- ============================================================================
+-- cleanup
+-- ============================================================================
+SELECT drop_graph('rls_vle', true);
+DROP OWNED BY rls_vle_user, rls_vle_user2, rls_vle_user3;
+DROP ROLE rls_vle_user, rls_vle_user2, rls_vle_user3;
diff --git a/src/backend/utils/adt/age_global_graph.c b/src/backend/utils/adt/age_global_graph.c
index 984d0fd01..44163bfbd 100644
--- a/src/backend/utils/adt/age_global_graph.c
+++ b/src/backend/utils/adt/age_global_graph.c
@@ -35,6 +35,13 @@
 #include "utils/memutils.h"
 #include "utils/snapmgr.h"
 #include "utils/builtins.h"
+#include "executor/spi.h"
+#include "lib/stringinfo.h"
+#include "miscadmin.h"
+#include "utils/rls.h"
+#include "utils/acl.h"
+#include "utils/inval.h"
+#include "utils/syscache.h"
 
 #if PG_VERSION_NUM >= 170000
 #include "storage/dsm_registry.h"
@@ -48,6 +55,7 @@
 #include "catalog/ag_graph.h"
 #include "catalog/ag_label.h"
 #include "utils/ag_cache.h"
+#include "utils/ag_guc.h"
 
 
 /* defines */
@@ -158,6 +166,12 @@ typedef struct GRAPH_global_context
     TransactionId xmin;            /* snapshot fallback: transaction xmin */
     TransactionId xmax;            /* snapshot fallback: transaction xmax */
     CommandId curcid;              /* snapshot fallback: command id */
+    Oid load_as_role;              /* role OID the cache was loaded as (RLS cache key) */
+    bool loaded_rls_enforced;      /* age.enforce_rls_in_traversal value at load */
+    bool vertices_rls_filtered;    /* true if any vertex label was RLS-filtered at load */
+    bool loaded_with_rls;          /* true if any label was loaded through RLS (SPI) */
+    bool security_invalidated;     /* set by ACL/RLS catalog inval callbacks */
+    List *label_table_oids;        /* label table OIDs, for targeted security inval */
     int64 num_loaded_vertices;     /* number of loaded vertices in this graph */
     int64 num_loaded_edges;        /* number of loaded edges in this graph */
     ListGraphId *vertices;         /* vertices for vertex hashtable cleanup */
@@ -220,6 +234,11 @@ static void create_GRAPH_global_hashtables(GRAPH_global_context *ggctx);
 static void load_GRAPH_global_hashtables(GRAPH_global_context *ggctx);
 static void load_vertex_hashtable(GRAPH_global_context *ggctx);
 static void load_edge_hashtable(GRAPH_global_context *ggctx);
+static void enforce_label_table_select_acl(Oid label_table_oid);
+static void register_ggctx_security_callbacks(void);
+static void ggctx_security_relcache_callback(Datum arg, Oid relid);
+static void ggctx_security_syscache_callback(Datum arg, int cacheid,
+                                             uint32 hashvalue);
 static void freeze_GRAPH_global_hashtables(GRAPH_global_context *ggctx);
 static List *get_ag_labels_names(Snapshot snapshot, Oid graph_oid,
                                  char label_type);
@@ -232,6 +251,12 @@ static bool insert_vertex_edge(GRAPH_global_context *ggctx,
 static bool insert_vertex_entry(GRAPH_global_context *ggctx, graphid vertex_id,
                                 Oid vertex_label_table_oid,
                                 ItemPointerData tid);
+static void load_vertex_label_rls(GRAPH_global_context *ggctx,
+                                  Oid vertex_label_table_oid,
+                                  char *vertex_label_name);
+static void load_edge_label_rls(GRAPH_global_context *ggctx,
+                                Oid edge_label_table_oid,
+                                char *edge_label_name);
 /* definitions */
 
 /*
@@ -247,6 +272,43 @@ static bool insert_vertex_entry(GRAPH_global_context *ggctx, graphid vertex_id,
  */
 bool is_ggctx_invalid(GRAPH_global_context *ggctx)
 {
+    /*
+     * Security-relevant cache keys, checked before the version-counter
+     * fast-path below. A cache that applied row-level security is never safe
+     * to reuse across statements: RLS policies can depend on session state
+     * (for example current_setting()) that is not part of the snapshot,
+     * role, or GUC keyed on here. A cache is also role- and GUC-specific
+     * because the SELECT ACL / RLS enforcement performed at load depends on
+     * the current role and on age.enforce_rls_in_traversal. Only graphs that
+     * actually enforced RLS pay the always-rebuild cost; ordinary graphs
+     * fall through to the normal version / snapshot validation below.
+     */
+    if (ggctx->loaded_with_rls)
+    {
+        return true;
+    }
+    if (ggctx->load_as_role != GetUserId() ||
+        ggctx->loaded_rls_enforced != age_enforce_rls_in_traversal)
+    {
+        return true;
+    }
+
+    /*
+     * If enforcement was on at load and a label table's ACL/RLS metadata (or
+     * the reading role's authorization) has since changed, an enforced cache
+     * may reflect stale permissions. The version-counter fast path below only
+     * tracks graph DATA changes, so GRANT/REVOKE, policy or RLS DDL, and role
+     * (BYPASSRLS / membership) changes would otherwise go unnoticed. The
+     * security-invalidation callbacks (register_ggctx_security_callbacks) set
+     * this flag on the relevant catalog invalidations; rebuild so the SELECT
+     * ACL / RLS state is re-evaluated. RLS-loaded caches already rebuild
+     * unconditionally above; non-enforced (GUC off) caches are unaffected.
+     */
+    if (ggctx->loaded_rls_enforced && ggctx->security_invalidated)
+    {
+        return true;
+    }
+
     /* use version counter if DSM or SHMEM mode is active */
     if (version_mode == VERSION_MODE_DSM || version_mode == VERSION_MODE_SHMEM)
     {
@@ -724,6 +786,279 @@ static bool insert_vertex_edge(GRAPH_global_context *ggctx,
     return false;
 }
 
+/*
+ * RLS-aware loader for a single vertex label table.
+ *
+ * Loads (id, ctid) through SPI so the planner applies row-level security
+ * policies (and table/column ACL) for the current role, then inserts each
+ * visible row into the graph's global vertex hashtable. The physical tuple
+ * location (ctid) is stored for the same lazy property fetch the direct-scan
+ * path uses (get_vertex_entry_properties); because only RLS-visible rows are
+ * loaded, that later heap_fetch only ever reads authorized tuples. Used in
+ * place of the direct heap scan when age.enforce_rls_in_traversal is on and RLS
+ * is active on the label table. Rows are fetched in batches so the entire label
+ * table is never materialized at once.
+ */
+static void load_vertex_label_rls(GRAPH_global_context *ggctx,
+                                  Oid vertex_label_table_oid,
+                                  char *vertex_label_name)
+{
+    MemoryContext ggctx_cxt = CurrentMemoryContext;
+    StringInfoData query;
+    SPIPlanPtr plan;
+    Portal portal;
+
+    if (SPI_connect() != SPI_OK_CONNECT)
+    {
+        ereport(ERROR,
+                (errcode(ERRCODE_INTERNAL_ERROR),
+                 errmsg("load_vertex_label_rls: SPI_connect failed")));
+    }
+
+    /*
+     * Build the query text AFTER SPI_connect(): SPI switches to its own
+     * procedure memory context here, so the StringInfo buffer and the
+     * quote_qualified_identifier() result are allocated in that context and
+     * freed by SPI_finish(), instead of leaking into the caller's long-lived
+     * (TopMemoryContext) cache-build context that is re-entered on every RLS
+     * cache rebuild.
+     */
+    initStringInfo(&query);
+    appendStringInfo(&query, "SELECT id, ctid FROM ONLY %s",
+                     quote_qualified_identifier(ggctx->graph_name,
+                                                vertex_label_name));
+
+    plan = SPI_prepare(query.data, 0, NULL);
+    if (plan == NULL)
+    {
+        ereport(ERROR,
+                (errcode(ERRCODE_INTERNAL_ERROR),
+                 errmsg("load_vertex_label_rls: SPI_prepare failed: %s",
+                        SPI_result_code_string(SPI_result))));
+    }
+
+    /* read-only cursor: reuses the active snapshot; RLS applied by planner */
+    portal = SPI_cursor_open(NULL, plan, NULL, NULL, true);
+
+    for (;;)
+    {
+        uint64 row;
+
+        SPI_cursor_fetch(portal, true, 10000);
+
+        if (SPI_processed == 0)
+        {
+            break;
+        }
+
+        for (row = 0; row < SPI_processed; row++)
+        {
+            HeapTuple tuple = SPI_tuptable->vals[row];
+            TupleDesc tupdesc = SPI_tuptable->tupdesc;
+            bool isnull = false;
+            graphid vertex_id;
+            ItemPointer tidptr;
+            ItemPointerData tid;
+            MemoryContext oldctx;
+
+            vertex_id = DatumGetInt64(SPI_getbinval(tuple, tupdesc, 1,
+                                                    &isnull));
+            tidptr = (ItemPointer) DatumGetPointer(SPI_getbinval(tuple, tupdesc,
+                                                                 2, &isnull));
+            /* ctid is never null for a scanned heap tuple; guard defensively */
+            if (isnull || tidptr == NULL)
+            {
+                continue;
+            }
+            tid = *tidptr;
+
+            /* build the entry in the cache's (persistent) memory context */
+            oldctx = MemoryContextSwitchTo(ggctx_cxt);
+            insert_vertex_entry(ggctx, vertex_id, vertex_label_table_oid, tid);
+            MemoryContextSwitchTo(oldctx);
+        }
+
+        SPI_freetuptable(SPI_tuptable);
+    }
+
+    SPI_cursor_close(portal);
+    SPI_finish();
+}
+
+/*
+ * RLS-aware loader for a single edge label table. See load_vertex_label_rls;
+ * this variant additionally loads start_id/end_id and populates the per-vertex
+ * edge lists, exactly like the direct-scan path, storing the ctid for the same
+ * lazy property fetch.
+ */
+static void load_edge_label_rls(GRAPH_global_context *ggctx,
+                                Oid edge_label_table_oid,
+                                char *edge_label_name)
+{
+    MemoryContext ggctx_cxt = CurrentMemoryContext;
+    StringInfoData query;
+    SPIPlanPtr plan;
+    Portal portal;
+
+    if (SPI_connect() != SPI_OK_CONNECT)
+    {
+        ereport(ERROR,
+                (errcode(ERRCODE_INTERNAL_ERROR),
+                 errmsg("load_edge_label_rls: SPI_connect failed")));
+    }
+
+    /*
+     * Build the query text AFTER SPI_connect(); see load_vertex_label_rls for
+     * why (SPI procedure memory context ownership avoids a per-rebuild leak).
+     */
+    initStringInfo(&query);
+    appendStringInfo(&query,
+                     "SELECT id, start_id, end_id, ctid FROM ONLY %s",
+                     quote_qualified_identifier(ggctx->graph_name,
+                                                edge_label_name));
+
+    plan = SPI_prepare(query.data, 0, NULL);
+    if (plan == NULL)
+    {
+        ereport(ERROR,
+                (errcode(ERRCODE_INTERNAL_ERROR),
+                 errmsg("load_edge_label_rls: SPI_prepare failed: %s",
+                        SPI_result_code_string(SPI_result))));
+    }
+
+    /* read-only cursor: reuses the active snapshot; RLS applied by planner */
+    portal = SPI_cursor_open(NULL, plan, NULL, NULL, true);
+
+    for (;;)
+    {
+        uint64 row;
+
+        SPI_cursor_fetch(portal, true, 10000);
+
+        if (SPI_processed == 0)
+        {
+            break;
+        }
+
+        for (row = 0; row < SPI_processed; row++)
+        {
+            HeapTuple tuple = SPI_tuptable->vals[row];
+            TupleDesc tupdesc = SPI_tuptable->tupdesc;
+            bool isnull = false;
+            graphid edge_id;
+            graphid edge_vertex_start_id;
+            graphid edge_vertex_end_id;
+            ItemPointer tidptr;
+            ItemPointerData tid;
+            MemoryContext oldctx;
+
+            edge_id = DatumGetInt64(SPI_getbinval(tuple, tupdesc, 1, &isnull));
+            edge_vertex_start_id = DatumGetInt64(SPI_getbinval(tuple, tupdesc,
+                                                               2, &isnull));
+            edge_vertex_end_id = DatumGetInt64(SPI_getbinval(tuple, tupdesc, 3,
+                                                             &isnull));
+            tidptr = (ItemPointer) DatumGetPointer(SPI_getbinval(tuple, tupdesc,
+                                                                 4, &isnull));
+            /* ctid is never null for a scanned heap tuple; guard defensively */
+            if (isnull || tidptr == NULL)
+            {
+                continue;
+            }
+            tid = *tidptr;
+
+            /*
+             * If a vertex label was RLS-filtered, drop any edge whose start or
+             * end vertex is not visible. This mirrors MATCH's inner-join
+             * behavior and avoids leaving a dangling edge in the cache.
+             */
+            if (ggctx->vertices_rls_filtered &&
+                (get_vertex_entry(ggctx, edge_vertex_start_id) == NULL ||
+                 get_vertex_entry(ggctx, edge_vertex_end_id) == NULL))
+            {
+                continue;
+            }
+
+            /* build the entries in the cache's (persistent) memory context */
+            oldctx = MemoryContextSwitchTo(ggctx_cxt);
+            insert_edge_entry(ggctx, edge_id, tid, edge_vertex_start_id,
+                              edge_vertex_end_id, edge_label_table_oid);
+            insert_vertex_edge(ggctx, edge_vertex_start_id,
+                               edge_vertex_end_id, edge_id, edge_label_name);
+            MemoryContextSwitchTo(oldctx);
+        }
+
+        SPI_freetuptable(SPI_tuptable);
+    }
+
+    SPI_cursor_close(portal);
+    SPI_finish();
+}
+
+/*
+ * Enforce table-level SELECT privilege on a graph label table before it is read
+ * by the low-level heap scans in the loaders below.
+ *
+ * Those scans bypass the executor, which is what normally enforces SELECT ACLs.
+ * Without this a role could read rows of a label table it has no SELECT
+ * privilege on through a VLE / traversal. The RLS half of that is handled
+ * separately by routing RLS-enabled labels through an SPI load
+ * (which the planner filters). This helper runs before both load paths: it is
+ * the only SELECT-ACL check on the raw fast path (used when RLS is not enabled),
+ * and a harmless early check before the SPI path (which the planner ACL-checks
+ * anyway). Gated by the same GUC as RLS enforcement so operators retain a
+ * single escape hatch.
+ */
+static void enforce_label_table_select_acl(Oid label_table_oid)
+{
+    if (!age_enforce_rls_in_traversal)
+    {
+        return;
+    }
+
+    /*
+     * table_open() / table_beginscan() below do not verify SELECT privilege -
+     * the executor normally does - so check it explicitly here.
+     *
+     * pg_class_aclcheck() only covers relation-level SELECT. PostgreSQL also
+     * allows column-level SELECT grants, which the executor (and therefore the
+     * RLS/SPI load path) honors. The loaders read every user column of the
+     * label table, so when the relation-level check fails, fall back to
+     * requiring SELECT on each live user column; this keeps the fast-path check
+     * consistent with what the executor would permit instead of over-denying a
+     * role that was granted the columns individually.
+     */
+    if (pg_class_aclcheck(label_table_oid, GetUserId(), ACL_SELECT) !=
+        ACLCHECK_OK)
+    {
+        Relation rel = table_open(label_table_oid, AccessShareLock);
+        TupleDesc tupdesc = RelationGetDescr(rel);
+        int i;
+
+        for (i = 0; i < tupdesc->natts; i++)
+        {
+            Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
+
+            /* skip dropped and system columns */
+            if (attr->attisdropped || attr->attnum <= 0)
+            {
+                continue;
+            }
+
+            if (pg_attribute_aclcheck(label_table_oid, attr->attnum,
+                                      GetUserId(), ACL_SELECT) != ACLCHECK_OK)
+            {
+                table_close(rel, AccessShareLock);
+                ereport(ERROR,
+                        (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+                         errmsg("permission denied for table %s",
+                                get_rel_name(label_table_oid))));
+            }
+        }
+
+        table_close(rel, AccessShareLock);
+    }
+}
+
 /* helper routine to load all vertices into the GRAPH global vertex hashtable */
 static void load_vertex_hashtable(GRAPH_global_context *ggctx)
 {
@@ -741,69 +1076,127 @@ static void load_vertex_hashtable(GRAPH_global_context *ggctx)
     /* get the names of all of the vertex label tables */
     vertex_label_names = get_ag_labels_names(snapshot, graph_oid,
                                              LABEL_TYPE_VERTEX);
-    /* go through all vertex label tables in list */
-    foreach (lc, vertex_label_names)
-    {
-        Relation graph_vertex_label;
-        TableScanDesc scan_desc;
-        HeapTuple tuple;
-        char *vertex_label_name;
-        Oid vertex_label_table_oid;
-        TupleDesc tupdesc;
-
-        /* get the vertex label name */
-        vertex_label_name = lfirst(lc);
-        /* get the vertex label name's OID */
-        vertex_label_table_oid = get_relname_relid(vertex_label_name,
-                                                   graph_namespace_oid);
-        /* open the relation (table) and begin the scan */
-        graph_vertex_label = table_open(vertex_label_table_oid, AccessShareLock);
-        scan_desc = table_beginscan(graph_vertex_label, snapshot, 0, NULL);
-        /* get the tupdesc - we don't need to release this one */
-        tupdesc = RelationGetDescr(graph_vertex_label);
-        /* bail if the number of columns differs */
-        if (tupdesc->natts != 2)
-        {
-            ereport(ERROR,
-                    (errcode(ERRCODE_UNDEFINED_TABLE),
-                     errmsg("Invalid number of attributes for %s.%s",
-                     ggctx->graph_name, vertex_label_name)));
-        }
-        /* get all tuples in table and insert them into graph hashtables */
-        while((tuple = heap_getnext(scan_desc, ForwardScanDirection)) != NULL)
+    /*
+     * The scan below enforces the SELECT ACL (and, when RLS is active on a
+     * label, loads it via SPI) and can therefore throw. get_ag_labels_names()
+     * built this list in TopMemoryContext (backend lifetime), so free it on
+     * the error path as well; otherwise a repeatedly-failing rebuild - for
+     * example a role retrying a VLE on a label it cannot read - leaks the
+     * list on every attempt.
+     */
+    PG_TRY();
+    {
+        /* go through all vertex label tables in list */
+        foreach (lc, vertex_label_names)
         {
-            graphid vertex_id;
-            bool inserted = false;
-
-            /* something is wrong if this isn't true */
-            if (!HeapTupleIsValid(tuple))
+            Relation graph_vertex_label;
+            TableScanDesc scan_desc;
+            HeapTuple tuple;
+            char *vertex_label_name;
+            Oid vertex_label_table_oid;
+            TupleDesc tupdesc;
+
+            /* get the vertex label name */
+            vertex_label_name = lfirst(lc);
+            /* get the vertex label name's OID */
+            vertex_label_table_oid = get_relname_relid(vertex_label_name,
+                                                       graph_namespace_oid);
+            /*
+             * Record this label table OID so the security-invalidation callbacks
+             * can target this graph when the label's ACL/RLS metadata changes.
+             */
+            ggctx->label_table_oids =
+                list_append_unique_oid(ggctx->label_table_oids,
+                                       vertex_label_table_oid);
+            /*
+             * Enforce SELECT privilege before any read of this label table
+             * (covers the raw heap-scan fast path below, which bypasses the
+             * executor's ACL check). No-op when the enforcement GUC is off.
+             */
+            enforce_label_table_select_acl(vertex_label_table_oid);
+            /*
+             * When RLS is active on this label for the current role, load it via
+             * SPI so the planner enforces the policies, then skip the direct
+             * scan below.
+             */
+            if (age_enforce_rls_in_traversal &&
+                check_enable_rls(vertex_label_table_oid, InvalidOid, true) ==
+                    RLS_ENABLED)
             {
-                elog(ERROR, "load_vertex_hashtable: !HeapTupleIsValid");
+                load_vertex_label_rls(ggctx, vertex_label_table_oid,
+                                      vertex_label_name);
+                /*
+                 * A vertex label was RLS-filtered, so the visible vertex set may
+                 * be a subset of what the edges reference. Remember this so the
+                 * edge loaders can drop edges whose endpoints are not visible.
+                 */
+                ggctx->vertices_rls_filtered = true;
+                ggctx->loaded_with_rls = true;
+                continue;
             }
-            Assert(HeapTupleIsValid(tuple));
+            /* open the relation (table) and begin the scan */
+            graph_vertex_label = table_open(vertex_label_table_oid, AccessShareLock);
+            scan_desc = table_beginscan(graph_vertex_label, snapshot, 0, NULL);
+            /* get the tupdesc - we don't need to release this one */
+            tupdesc = RelationGetDescr(graph_vertex_label);
+            /* bail if the number of columns differs */
+            if (tupdesc->natts != 2)
+            {
+                ereport(ERROR,
+                        (errcode(ERRCODE_UNDEFINED_TABLE),
+                         errmsg("Invalid number of attributes for %s.%s",
+                         ggctx->graph_name, vertex_label_name)));
+            }
+            /* get all tuples in table and insert them into graph hashtables */
+            while((tuple = heap_getnext(scan_desc, ForwardScanDirection)) != NULL)
+            {
+                graphid vertex_id;
+                bool inserted = false;
 
-            /* get the vertex id */
-            vertex_id = DatumGetInt64(column_get_datum(tupdesc, tuple, 0, "id",
-                                                       GRAPHIDOID, true));
+                /* something is wrong if this isn't true */
+                if (!HeapTupleIsValid(tuple))
+                {
+                    elog(ERROR, "load_vertex_hashtable: !HeapTupleIsValid");
+                }
+                Assert(HeapTupleIsValid(tuple));
 
-            /* insert vertex into vertex hashtable with TID (no property copy) */
-            inserted = insert_vertex_entry(ggctx, vertex_id,
-                                           vertex_label_table_oid,
-                                           tuple->t_self);
+                /* get the vertex id */
+                vertex_id = DatumGetInt64(column_get_datum(tupdesc, tuple, 0, "id",
+                                                           GRAPHIDOID, true));
 
-            /* warn if there is a duplicate */
-            if (!inserted)
-            {
-                 ereport(WARNING,
-                         (errcode(ERRCODE_DATA_EXCEPTION),
-                          errmsg("ignored duplicate vertex")));
+                /* insert vertex into vertex hashtable with TID (no property copy) */
+                inserted = insert_vertex_entry(ggctx, vertex_id,
+                                               vertex_label_table_oid,
+                                               tuple->t_self);
+
+                /* warn if there is a duplicate */
+                if (!inserted)
+                {
+                     ereport(WARNING,
+                             (errcode(ERRCODE_DATA_EXCEPTION),
+                              errmsg("ignored duplicate vertex")));
+                }
             }
-        }
 
-        /* end the scan and close the relation */
-        table_endscan(scan_desc);
-        table_close(graph_vertex_label, AccessShareLock);
+            /* end the scan and close the relation */
+            table_endscan(scan_desc);
+            table_close(graph_vertex_label, AccessShareLock);
+        }
     }
+    PG_CATCH();
+    {
+        list_free_deep(vertex_label_names);
+        PG_RE_THROW();
+    }
+    PG_END_TRY();
+
+    /*
+     * Free the transient list of label names (and the palloc'd Name entries)
+     * returned by get_ag_labels_names(). The global graph context is built in
+     * TopMemoryContext, so this list would otherwise persist for the life of
+     * the backend and accumulate on every rebuild.
+     */
+    list_free_deep(vertex_label_names);
 }
 
 /*
@@ -843,94 +1236,157 @@ static void load_edge_hashtable(GRAPH_global_context *ggctx)
     /* get the names of all of the edge label tables */
     edge_label_names = get_ag_labels_names(snapshot, graph_oid,
                                            LABEL_TYPE_EDGE);
-    /* go through all edge label tables in list */
-    foreach (lc, edge_label_names)
-    {
-        Relation graph_edge_label;
-        TableScanDesc scan_desc;
-        HeapTuple tuple;
-        char *edge_label_name;
-        Oid edge_label_table_oid;
-        TupleDesc tupdesc;
-
-        /* get the edge label name */
-        edge_label_name = lfirst(lc);
-        /* get the edge label name's OID */
-        edge_label_table_oid = get_relname_relid(edge_label_name,
-                                                 graph_namespace_oid);
-        /* open the relation (table) and begin the scan */
-        graph_edge_label = table_open(edge_label_table_oid, AccessShareLock);
-        scan_desc = table_beginscan(graph_edge_label, snapshot, 0, NULL);
-        /* get the tupdesc - we don't need to release this one */
-        tupdesc = RelationGetDescr(graph_edge_label);
-        /* bail if the number of columns differs */
-        if (tupdesc->natts != 4)
-        {
-            ereport(ERROR,
-                    (errcode(ERRCODE_UNDEFINED_TABLE),
-                     errmsg("Invalid number of attributes for %s.%s",
-                     ggctx->graph_name, edge_label_name)));
-        }
-        /* get all tuples in table and insert them into graph hashtables */
-        while((tuple = heap_getnext(scan_desc, ForwardScanDirection)) != NULL)
+    /*
+     * The scan below enforces the SELECT ACL (and, when RLS is active on a
+     * label, loads it via SPI) and can therefore throw. get_ag_labels_names()
+     * built this list in TopMemoryContext (backend lifetime), so free it on
+     * the error path as well; otherwise a repeatedly-failing rebuild - for
+     * example a role retrying a VLE on a label it cannot read - leaks the
+     * list on every attempt.
+     */
+    PG_TRY();
+    {
+        /* go through all edge label tables in list */
+        foreach (lc, edge_label_names)
         {
-            graphid edge_id;
-            graphid edge_vertex_start_id;
-            graphid edge_vertex_end_id;
-            bool inserted = false;
-
-            /* something is wrong if this isn't true */
-            if (!HeapTupleIsValid(tuple))
+            Relation graph_edge_label;
+            TableScanDesc scan_desc;
+            HeapTuple tuple;
+            char *edge_label_name;
+            Oid edge_label_table_oid;
+            TupleDesc tupdesc;
+
+            /* get the edge label name */
+            edge_label_name = lfirst(lc);
+            /* get the edge label name's OID */
+            edge_label_table_oid = get_relname_relid(edge_label_name,
+                                                     graph_namespace_oid);
+            /*
+             * Record this label table OID so the security-invalidation callbacks
+             * can target this graph when the label's ACL/RLS metadata changes.
+             */
+            ggctx->label_table_oids =
+                list_append_unique_oid(ggctx->label_table_oids,
+                                       edge_label_table_oid);
+            /*
+             * Enforce SELECT privilege before any read of this label table
+             * (covers the raw heap-scan fast path below, which bypasses the
+             * executor's ACL check). No-op when the enforcement GUC is off.
+             */
+            enforce_label_table_select_acl(edge_label_table_oid);
+            /*
+             * When RLS is active on this label for the current role, load it via
+             * SPI so the planner enforces the policies, then skip the direct
+             * scan below.
+             */
+            if (age_enforce_rls_in_traversal &&
+                check_enable_rls(edge_label_table_oid, InvalidOid, true) ==
+                    RLS_ENABLED)
             {
-                elog(ERROR, "load_edge_hashtable: !HeapTupleIsValid");
+                load_edge_label_rls(ggctx, edge_label_table_oid, edge_label_name);
+                ggctx->loaded_with_rls = true;
+                continue;
             }
-            Assert(HeapTupleIsValid(tuple));
-
-            /* get the edge id */
-            edge_id = DatumGetInt64(column_get_datum(tupdesc, tuple, 0, "id",
-                                                     GRAPHIDOID, true));
-            /* get the edge start_id (start vertex id) */
-            edge_vertex_start_id = DatumGetInt64(column_get_datum(tupdesc,
-                                                                  tuple, 1,
-                                                                  "start_id",
-                                                                  GRAPHIDOID,
-                                                                  true));
-            /* get the edge end_id (end vertex id)*/
-            edge_vertex_end_id = DatumGetInt64(column_get_datum(tupdesc, tuple,
-                                                                2, "end_id",
-                                                                GRAPHIDOID,
-                                                                true));
-
-            /* insert edge into edge hashtable with TID (no property copy) */
-            inserted = insert_edge_entry(ggctx, edge_id, tuple->t_self,
-                                         edge_vertex_start_id,
-                                         edge_vertex_end_id,
-                                         edge_label_table_oid);
-
-            /* warn if there is a duplicate */
-            if (!inserted)
+            /* open the relation (table) and begin the scan */
+            graph_edge_label = table_open(edge_label_table_oid, AccessShareLock);
+            scan_desc = table_beginscan(graph_edge_label, snapshot, 0, NULL);
+            /* get the tupdesc - we don't need to release this one */
+            tupdesc = RelationGetDescr(graph_edge_label);
+            /* bail if the number of columns differs */
+            if (tupdesc->natts != 4)
             {
-                 ereport(WARNING,
-                         (errcode(ERRCODE_DATA_EXCEPTION),
-                          errmsg("ignored duplicate edge")));
+                ereport(ERROR,
+                        (errcode(ERRCODE_UNDEFINED_TABLE),
+                         errmsg("Invalid number of attributes for %s.%s",
+                         ggctx->graph_name, edge_label_name)));
             }
-
-            /* insert the edge into the start and end vertices edge lists */
-            inserted = insert_vertex_edge(ggctx, edge_vertex_start_id,
-                                          edge_vertex_end_id, edge_id,
-                                          edge_label_name);
-            if (!inserted)
+            /* get all tuples in table and insert them into graph hashtables */
+            while((tuple = heap_getnext(scan_desc, ForwardScanDirection)) != NULL)
             {
-                 ereport(WARNING,
-                         (errcode(ERRCODE_DATA_EXCEPTION),
-                          errmsg("ignored malformed or dangling edge")));
+                graphid edge_id;
+                graphid edge_vertex_start_id;
+                graphid edge_vertex_end_id;
+                bool inserted = false;
+
+                /* something is wrong if this isn't true */
+                if (!HeapTupleIsValid(tuple))
+                {
+                    elog(ERROR, "load_edge_hashtable: !HeapTupleIsValid");
+                }
+                Assert(HeapTupleIsValid(tuple));
+
+                /* get the edge id */
+                edge_id = DatumGetInt64(column_get_datum(tupdesc, tuple, 0, "id",
+                                                         GRAPHIDOID, true));
+                /* get the edge start_id (start vertex id) */
+                edge_vertex_start_id = DatumGetInt64(column_get_datum(tupdesc,
+                                                                      tuple, 1,
+                                                                      "start_id",
+                                                                      GRAPHIDOID,
+                                                                      true));
+                /* get the edge end_id (end vertex id)*/
+                edge_vertex_end_id = DatumGetInt64(column_get_datum(tupdesc, tuple,
+                                                                    2, "end_id",
+                                                                    GRAPHIDOID,
+                                                                    true));
+
+                /*
+                 * If a vertex label was RLS-filtered, drop any edge whose start
+                 * or end vertex is not visible (mirrors MATCH's inner-join
+                 * behavior).
+                 */
+                if (ggctx->vertices_rls_filtered &&
+                    (get_vertex_entry(ggctx, edge_vertex_start_id) == NULL ||
+                     get_vertex_entry(ggctx, edge_vertex_end_id) == NULL))
+                {
+                    continue;
+                }
+
+                /* insert edge into edge hashtable with TID (no property copy) */
+                inserted = insert_edge_entry(ggctx, edge_id, tuple->t_self,
+                                             edge_vertex_start_id,
+                                             edge_vertex_end_id,
+                                             edge_label_table_oid);
+
+                /* warn if there is a duplicate */
+                if (!inserted)
+                {
+                     ereport(WARNING,
+                             (errcode(ERRCODE_DATA_EXCEPTION),
+                              errmsg("ignored duplicate edge")));
+                }
+
+                /* insert the edge into the start and end vertices edge lists */
+                inserted = insert_vertex_edge(ggctx, edge_vertex_start_id,
+                                              edge_vertex_end_id, edge_id,
+                                              edge_label_name);
+                if (!inserted)
+                {
+                     ereport(WARNING,
+                             (errcode(ERRCODE_DATA_EXCEPTION),
+                              errmsg("ignored malformed or dangling edge")));
+                }
             }
-        }
 
-        /* end the scan and close the relation */
-        table_endscan(scan_desc);
-        table_close(graph_edge_label, AccessShareLock);
+            /* end the scan and close the relation */
+            table_endscan(scan_desc);
+            table_close(graph_edge_label, AccessShareLock);
+        }
+    }
+    PG_CATCH();
+    {
+        list_free_deep(edge_label_names);
+        PG_RE_THROW();
     }
+    PG_END_TRY();
+
+    /*
+     * Free the transient list of label names (and the palloc'd Name entries)
+     * returned by get_ag_labels_names(). The global graph context is built in
+     * TopMemoryContext, so this list would otherwise persist for the life of
+     * the backend and accumulate on every rebuild.
+     */
+    list_free_deep(edge_label_names);
 }
 
 /*
@@ -962,6 +1418,10 @@ static bool free_specific_GRAPH_global_context(GRAPH_global_context *ggctx)
     pfree_if_not_null(ggctx->graph_name);
     ggctx->graph_name = NULL;
 
+    /* free the label table OID list used for security invalidation */
+    list_free(ggctx->label_table_oids);
+    ggctx->label_table_oids = NULL;
+
     ggctx->graph_oid = InvalidOid;
     ggctx->next = NULL;
 
@@ -1025,14 +1485,85 @@ static bool free_specific_GRAPH_global_context(GRAPH_global_context *ggctx)
     return true;
 }
 
+/*
+ * Security-invalidation callbacks for the global-graph cache.
+ *
+ * The version-counter fast path in is_ggctx_invalid only tracks graph DATA
+ * changes, so a cache built under enforcement (direct scan + SELECT ACL, no RLS)
+ * would keep serving rows after a GRANT/REVOKE, a policy/RLS DDL change, or a
+ * role authorization change, because none of those bump the data version. These
+ * callbacks mark such caches stale on the relevant catalog invalidations;
+ * is_ggctx_invalid then rebuilds and re-evaluates authorization.
+ */
+
+/*
+ * Relcache invalidation. Table-level (pg_class) and column-level (pg_attribute)
+ * SELECT ACL changes, CREATE/ALTER/DROP POLICY, and ENABLE/DISABLE/FORCE ROW
+ * LEVEL SECURITY all invalidate the affected relation's relcache. Mark every
+ * cached graph whose label tables include relid; InvalidOid is a global reset.
+ */
+static void ggctx_security_relcache_callback(Datum arg, Oid relid)
+{
+    GRAPH_global_context *ggctx;
+
+    for (ggctx = global_graph_contexts; ggctx != NULL; ggctx = ggctx->next)
+    {
+        if (!OidIsValid(relid) ||
+            list_member_oid(ggctx->label_table_oids, relid))
+        {
+            ggctx->security_invalidated = true;
+        }
+    }
+}
+
+/*
+ * Syscache invalidation on pg_authid (AUTHOID) and pg_auth_members
+ * (AUTHMEMMEMROLE / AUTHMEMROLEMEM). Role attribute (for example BYPASSRLS) and
+ * membership changes affect authorization but carry no relid we can map to a
+ * label table, so mark all cached contexts stale. Such DDL is rare, so the
+ * broad invalidation is inexpensive.
+ */
+static void ggctx_security_syscache_callback(Datum arg, int cacheid,
+                                             uint32 hashvalue)
+{
+    GRAPH_global_context *ggctx;
+
+    for (ggctx = global_graph_contexts; ggctx != NULL; ggctx = ggctx->next)
+    {
+        ggctx->security_invalidated = true;
+    }
+}
+
+/*
+ * Register the security-invalidation callbacks once per backend. Registered
+ * lazily from manage_GRAPH_global_contexts, before the first cache is built, so
+ * there is never a cache in existence that could miss an earlier invalidation.
+ */
+static void register_ggctx_security_callbacks(void)
+{
+    static bool registered = false;
+
+    if (registered)
+    {
+        return;
+    }
+
+    CacheRegisterRelcacheCallback(ggctx_security_relcache_callback, (Datum) 0);
+    CacheRegisterSyscacheCallback(AUTHOID, ggctx_security_syscache_callback,
+                                  (Datum) 0);
+    CacheRegisterSyscacheCallback(AUTHMEMMEMROLE,
+                                  ggctx_security_syscache_callback, (Datum) 0);
+    CacheRegisterSyscacheCallback(AUTHMEMROLEMEM,
+                                  ggctx_security_syscache_callback, (Datum) 0);
+
+    registered = true;
+}
+
 /*
  * Helper function to manage the GRAPH global contexts. It will create the
  * context for the graph specified, provided it isn't already built and valid.
  * During processing it will free (delete) all invalid GRAPH contexts. It
  * returns the GRAPH global context for the specified graph.
- *
- * NOTE: Function uses a MUTEX for global_graph_contexts
- *
  */
 GRAPH_global_context *manage_GRAPH_global_contexts(char *graph_name,
                                                    Oid graph_oid)
@@ -1045,6 +1576,14 @@ GRAPH_global_context *manage_GRAPH_global_contexts(char *graph_name,
     /* we need a higher context, or one that isn't destroyed by SRF exit */
     oldctx = MemoryContextSwitchTo(TopMemoryContext);
 
+    /*
+     * Ensure the ACL/RLS security-invalidation callbacks are registered (once
+     * per backend) before any cache is built, so the version-counter fast path
+     * in is_ggctx_invalid cannot serve stale permissions after GRANT/REVOKE,
+     * policy/RLS DDL, or role authorization changes.
+     */
+    register_ggctx_security_callbacks();
+
     /*
      * We need to see if any GRAPH global contexts already exist and if any do
      * for this particular graph. There are 5 possibilities -
@@ -1118,20 +1657,23 @@ GRAPH_global_context *manage_GRAPH_global_contexts(char *graph_name,
         curr_ggctx = curr_ggctx->next;
     }
 
-    /* otherwise, we need to create one and possibly attach it */
+    /*
+     * Otherwise we need to build a new context for this graph.
+     *
+     * Build it DETACHED from the shared list, then attach it after a
+     * successful build. The RLS-aware load path runs an SPI query that
+     * evaluates user-defined row-level security policy expressions, and such
+     * an expression can call back into AGE (for example a policy that invokes
+     * a function performing its own VLE / global-graph traversal), re-entering
+     * this function on the same backend. If the half-built context were
+     * already on the shared list, that re-entrant call could find and use it
+     * before it was fully loaded. Building detached avoids that; the
+     * re-entrant call builds its own context and the duplicate is discarded
+     * on attach below. (This code path holds no lock: the context list is
+     * per-backend.)
+     */
     new_ggctx = palloc0(sizeof(GRAPH_global_context));
-
-    if (global_graph_contexts != NULL)
-    {
-        new_ggctx->next = global_graph_contexts;
-    }
-    else
-    {
-        new_ggctx->next = NULL;
-    }
-
-    /* set the global context variable */
-    global_graph_contexts = new_ggctx;
+    new_ggctx->next = NULL;
 
     /* set the graph name and oid */
     new_ggctx->graph_name = pstrdup(graph_name);
@@ -1145,14 +1687,101 @@ GRAPH_global_context *manage_GRAPH_global_contexts(char *graph_name,
     new_ggctx->xmax = GetActiveSnapshot()->xmax;
     new_ggctx->curcid = GetActiveSnapshot()->curcid;
 
+    /*
+     * Record the role and enforcement GUC the cache is loaded under. The
+     * SELECT ACL / RLS enforcement applied during load is role- and
+     * GUC-dependent, so a cache built as one role (or GUC state) must not be
+     * reused for another (see is_ggctx_invalid).
+     */
+    new_ggctx->load_as_role = GetUserId();
+    new_ggctx->loaded_rls_enforced = age_enforce_rls_in_traversal;
+
     /* initialize our vertices list */
     new_ggctx->vertices = NULL;
 
-    /* build the hashtables for this graph */
-    create_GRAPH_global_hashtables(new_ggctx);
-    load_GRAPH_global_hashtables(new_ggctx);
-    freeze_GRAPH_global_hashtables(new_ggctx);
+    /* build the hashtables for this graph (detached from the shared list) */
+    PG_TRY();
+    {
+        create_GRAPH_global_hashtables(new_ggctx);
+        load_GRAPH_global_hashtables(new_ggctx);
+        freeze_GRAPH_global_hashtables(new_ggctx);
+    }
+    PG_CATCH();
+    {
+        /*
+         * The context is not attached to the shared list, so on a failed
+         * load (for example an RLS or ACL error) just free the
+         * partially-built context and re-throw. Its allocations live in
+         * TopMemoryContext (backend lifetime), so without this a role that
+         * repeatedly triggers a load failure - for example retrying a VLE on
+         * a table it cannot read - would leak memory until the backend exits.
+         * free_specific_GRAPH_global_context is safe on a partially-built
+         * context: the vertices list and vertex_hashtable are kept in sync by
+         * insert_vertex_entry, the edge_table lives in its own memory context,
+         * and it only uses pfree / hash_search / hash_destroy /
+         * MemoryContextDelete, none of which re-throw.
+         */
+        (void) free_specific_GRAPH_global_context(new_ggctx);
+        MemoryContextSwitchTo(oldctx);
+        PG_RE_THROW();
+    }
+    PG_END_TRY();
+
+    /*
+     * Attach. While we were building, a re-entrant call from an RLS policy on
+     * this same backend may already have created and attached a context for
+     * this graph. Reuse it only when it is still valid for the current
+     * security context - same role, same enforcement GUC, and not loaded under
+     * RLS - as tested by is_ggctx_invalid(). A re-entrant traversal reached
+     * through a SECURITY DEFINER RLS policy can attach a context loaded as a
+     * different role; returning it here would expose rows the current role may
+     * not be allowed to see. If the existing context is not reusable, detach
+     * and discard it so the list keeps a single, correct context per graph.
+     */
+    prev_ggctx = NULL;
+    curr_ggctx = global_graph_contexts;
+    while (curr_ggctx != NULL)
+    {
+        if (curr_ggctx->graph_oid == graph_oid)
+        {
+            if (!is_ggctx_invalid(curr_ggctx))
+            {
+                (void) free_specific_GRAPH_global_context(new_ggctx);
+                MemoryContextSwitchTo(oldctx);
+                return curr_ggctx;
+            }
+
+            /* detach the incompatible context and discard it */
+            if (prev_ggctx == NULL)
+            {
+                global_graph_contexts = curr_ggctx->next;
+            }
+            else
+            {
+                prev_ggctx->next = curr_ggctx->next;
+            }
+            if (!free_specific_GRAPH_global_context(curr_ggctx))
+            {
+                /*
+                 * The detached context was internally inconsistent (a
+                 * vertex or edge entry went missing). Free our freshly-
+                 * built context first so it does not leak, then surface
+                 * the corruption as the other cleanup paths in this file
+                 * do.
+                 */
+                (void) free_specific_GRAPH_global_context(new_ggctx);
+                ereport(ERROR, (errcode(ERRCODE_DATA_EXCEPTION),
+                                errmsg("missing vertex or edge entry during free")));
+            }
+            break;
+        }
+        prev_ggctx = curr_ggctx;
+        curr_ggctx = curr_ggctx->next;
+    }
 
+    /* prepend our freshly-built context to the shared list */
+    new_ggctx->next = global_graph_contexts;
+    global_graph_contexts = new_ggctx;
 
     /* switch back to the previous memory context */
     MemoryContextSwitchTo(oldctx);
diff --git a/src/backend/utils/ag_guc.c b/src/backend/utils/ag_guc.c
index 86b4e00bc..15754d8c2 100644
--- a/src/backend/utils/ag_guc.c
+++ b/src/backend/utils/ag_guc.c
@@ -23,6 +23,7 @@
 #include "utils/ag_guc.h"
 
 bool age_enable_containment = true;
+bool age_enforce_rls_in_traversal = true;
 
 /*
  * Defines AGE's custom configuration parameters.
@@ -42,5 +43,15 @@ void define_config_params(void)
                              NULL,
                              NULL,
                              NULL);
+    DefineCustomBoolVariable("age.enforce_rls_in_traversal",
+                             "Enforce row-level security during VLE / global-graph traversal. When off, the traversal cache is loaded with a direct scan that bypasses RLS.",
+                             NULL,
+                             &age_enforce_rls_in_traversal,
+                             true,
+                             PGC_SUSET,
+                             0,
+                             NULL,
+                             NULL,
+                             NULL);
     EmitWarningsOnPlaceholders("age");
 }
diff --git a/src/include/utils/ag_guc.h b/src/include/utils/ag_guc.h
index 52fab2b85..c0f34795b 100644
--- a/src/include/utils/ag_guc.h
+++ b/src/include/utils/ag_guc.h
@@ -39,6 +39,15 @@
  */
 extern bool age_enable_containment;
 
+/*
+ * If set true (the default), variable-length edge (VLE) traversal and other
+ * global-graph readers enforce row-level security (RLS) policies (and
+ * table/column ACL) on the graph label tables. When false, the global graph
+ * cache is loaded with a direct heap scan that bypasses RLS (legacy behavior:
+ * faster, but not RLS-aware).
+ */
+extern bool age_enforce_rls_in_traversal;
+
 void define_config_params(void);
 
 #endif