fix(grants): order READ before WRITE on bulk stage grants

snowcap/blueprint.py

@@ -1525,14 +1525,20 @@ def _create_grandparent_refs(self) -> None:
                 resource.requires(resource.container.container)
 
     def _create_stage_privilege_refs(self) -> None:
-        stage_grants: dict[str, list[Grant]] = {}
+        stage_grants: dict[tuple, list[Grant]] = {}
 
         for resource in _walk(self._root):
             if isinstance(resource, Grant):
-                if resource._data.on_type == ResourceType.STAGE:
-                    if resource._data.on not in stage_grants:
-                        stage_grants[resource._data.on] = []
-                    stage_grants[resource._data.on].append(resource)
+                # Snowflake requires READ before/with WRITE on a stage. Catch
+                # both direct grants (on_type == STAGE) and bulk ALL/FUTURE
+                # grants (items_type == STAGE), keyed by exact scope so the
+                # WRITE -> READ dependency below covers each case.
+                d = resource._data
+                if d.on_type == ResourceType.STAGE or d.items_type == ResourceType.STAGE:
+                    key = (d.on_type, d.on, d.grant_type, d.items_type)
+                    if key not in stage_grants:
+                        stage_grants[key] = []
+                    stage_grants[key].append(resource)
 
         def _apply_refs(stage_grants):
             for stage in stage_grants.keys():

tests/integration/test_blueprint.py

@@ -637,6 +637,42 @@ def test_stage_read_write_privilege_execution_order(cursor, suffix, marked_for_c
     blueprint.apply(session, plan)
 
 
+def test_bulk_stage_read_write_privilege_execution_order(cursor, suffix, test_db, marked_for_cleanup):
+    """
+    Snowflake rejects WRITE on a stage unless READ is granted first/together.
+    This also covers bulk ALL/FUTURE stage grants: without the WRITE -> READ
+    dependency, same-level grants for a role run concurrently and WRITE can
+    reach Snowflake before READ, failing with a privilege order violation.
+
+    Apply READ/WRITE across all four ALL/FUTURE x DATABASE/SCHEMA scopes and
+    assert the apply succeeds (it raises pre-fix).
+    """
+    session = cursor.connection
+
+    role_name = f"BULK_STAGE_ACCESS_ROLE_{suffix}"
+    role = res.Role(name=role_name)
+
+    scopes = [
+        f"all stages in database {test_db}",
+        f"all stages in schema {test_db}.PUBLIC",
+        f"future stages in database {test_db}",
+        f"future stages in schema {test_db}.PUBLIC",
+    ]
+
+    grants = []
+    for scope in scopes:
+        grants.append(res.Grant(priv="READ", on=scope, to=role))
+        grants.append(res.Grant(priv="WRITE", on=scope, to=role))
+
+    blueprint = Blueprint(resources=[role, *grants])
+
+    marked_for_cleanup.append(role)
+
+    plan = blueprint.plan(session)
+    assert len(plan) == 1 + len(grants)
+    blueprint.apply(session, plan)
+
+
 def test_grant_database_role_to_database_role(cursor, suffix, marked_for_cleanup):
     session = cursor.connection
     bp = Blueprint()

tests/test_blueprint.py

@@ -442,6 +442,56 @@ def test_blueprint_reference_sorting(session_ctx, remote_state):
     assert db3_change.resource_cls == res.Database
 
 
+def test_blueprint_bulk_stage_read_write_ordering(session_ctx):
+    """
+    Bulk (ALL/FUTURE) stage grants must order READ before WRITE, just like
+    grants on a single named stage. _create_stage_privilege_refs should make
+    each WRITE grant depend on the matching READ grant over the same scope,
+    for every ALL/FUTURE x DATABASE/SCHEMA combination.
+    """
+    role = res.Role(name="R_RW")
+
+    # Every ALL/FUTURE x DATABASE/SCHEMA combination must order READ before
+    # WRITE, since _create_stage_privilege_refs keys on items_type == STAGE
+    # regardless of container type or grant type.
+    all_db_read = res.Grant(priv="READ", on="all stages in database DB", to=role)
+    all_db_write = res.Grant(priv="WRITE", on="all stages in database DB", to=role)
+    all_sc_read = res.Grant(priv="READ", on="all stages in schema DB.SC", to=role)
+    all_sc_write = res.Grant(priv="WRITE", on="all stages in schema DB.SC", to=role)
+    future_db_read = res.Grant(priv="READ", on="future stages in database DB", to=role)
+    future_db_write = res.Grant(priv="WRITE", on="future stages in database DB", to=role)
+    future_sc_read = res.Grant(priv="READ", on="future stages in schema DB.SC", to=role)
+    future_sc_write = res.Grant(priv="WRITE", on="future stages in schema DB.SC", to=role)
+
+    blueprint = Blueprint(
+        resources=[
+            role,
+            all_db_read,
+            all_db_write,
+            all_sc_read,
+            all_sc_write,
+            future_db_read,
+            future_db_write,
+            future_sc_read,
+            future_sc_write,
+        ]
+    )
+    blueprint._finalize(session_ctx)
+
+    # WRITE depends on READ -> READ is applied first, for each scope
+    assert all_db_read in all_db_write.refs
+    assert all_sc_read in all_sc_write.refs
+    assert future_db_read in future_db_write.refs
+    assert future_sc_read in future_sc_write.refs
+
+    # Scopes must not cross-link: a WRITE only depends on the READ over its
+    # exact same scope, not on READs from other container/grant-type scopes.
+    assert all_sc_read not in all_db_write.refs
+    assert future_db_read not in all_db_write.refs
+    assert future_sc_read not in future_db_write.refs
+    assert all_db_read not in future_sc_write.refs
+
+
 def test_blueprint_ownership_sorting(session_ctx, remote_state):
 
     role = res.Role(name="SOME_ROLE")