From 609152fb6fdaa139eb1709c8e2c0a1455bc5385b Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 00:21:15 -0700
Subject: [PATCH 01/10] Bump cube version if any upstreams changed

---
 .../datajunction_server/internal/nodes.py     | 85 ++++++++++++++++---
 datajunction-server/tests/api/nodes_test.py   | 45 ++++++++++
 2 files changed, 120 insertions(+), 10 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index a0c45d3d4..0894f0043 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1701,29 +1701,32 @@ async def _propagate_update_downstream(
     - new columns: won't affect downstream nodes
     """
     _logger.info("Propagating update of node %s downstream", node.name)
-    downstreams = await get_downstream_nodes(
+    all_downstreams = await get_downstream_nodes(
         session,
         node.name,
         include_deactivated=False,
-        include_cubes=False,
+        include_cubes=True,
+    )
+    non_cube_downstreams = topological_sort(
+        [n for n in all_downstreams if n.type != NodeType.CUBE],
     )
-    downstreams = topological_sort(downstreams)
+    cube_downstreams = [n for n in all_downstreams if n.type == NodeType.CUBE]
     _logger.info(
-        "Node %s updated — revalidating %s downstreams",
+        "Node %s updated — revalidating %s downstreams, bumping %s cubes",
         node.name,
-        len(downstreams),
+        len(non_cube_downstreams),
+        len(cube_downstreams),
     )
 
-    # The downstreams need to be sorted topologically in order for the updates to be done
-    # in the right order. Otherwise it is possible for a leaf node like a metric to be updated
-    # before its upstreams are updated.
-    for idx, downstream in enumerate(downstreams):
+    # Revalidate non-cube downstreams in topological order so parents are
+    # processed before their children.
+    for idx, downstream in enumerate(non_cube_downstreams):
         original_node_revision = downstream.current
         previous_status = original_node_revision.status
         _logger.info(
             "[%s/%s] Revalidating downstream %s due to update of node %s",
             idx + 1,
-            len(downstreams),
+            len(non_cube_downstreams),
             downstream.name,
             node.name,
         )
@@ -1785,6 +1788,68 @@ async def _propagate_update_downstream(
             )
         await session.commit()
 
+    # A cube's element list doesn't change when an upstream metric's SQL
+    # changes, so update_cube_node's diff logic won't fire. Create a new
+    # minor revision so callers that gate on version ID see the change.
+    for cube in cube_downstreams:
+        cube_node = await Node.get_cube_by_name(session, cube.name)
+        if not cube_node or not cube_node.current:
+            continue
+        current_rev = cube_node.current
+
+        create_cube = CreateCubeNode(
+            name=current_rev.name,
+            display_name=current_rev.display_name,
+            description=current_rev.description,
+            metrics=[m.name for m in current_rev.cube_metrics()],
+            dimensions=current_rev.cube_dimensions(),
+            mode=current_rev.mode,
+            filters=current_rev.cube_filters or [],
+            custom_metadata=current_rev.custom_metadata,
+        )
+
+        with session.no_autoflush:
+            new_cube_revision = await create_cube_node_revision(
+                session,
+                create_cube,
+                current_user,
+            )
+            old_version = Version.parse(current_rev.version)
+            new_cube_revision.version = str(old_version.next_minor_version())
+            new_cube_revision.node = current_rev.node
+            new_cube_revision.node.current_version = new_cube_revision.version
+
+            await save_history(
+                event=History(
+                    entity_type=EntityType.NODE,
+                    entity_name=new_cube_revision.name,
+                    node=new_cube_revision.name,
+                    activity_type=ActivityType.UPDATE,
+                    details={
+                        "version": new_cube_revision.version,
+                        "upstream": {
+                            "node": node.name,
+                            "version": node.current_version,
+                        },
+                        "reason": f"Caused by update of `{node.name}` to "
+                        f"{node.current_version}",
+                    },
+                    pre={"version": current_rev.version},
+                    post={"version": new_cube_revision.version},
+                    user=current_user.username,
+                ),
+                session=session,
+            )
+
+        await session.commit()
+        _logger.info(
+            "Bumped cube %s from %s to %s due to upstream change of %s",
+            cube.name,
+            current_rev.version,
+            new_cube_revision.version,
+            node.name,
+        )
+
 
 def copy_existing_node_revision(old_revision: NodeRevision, current_user: User):
     """
diff --git a/datajunction-server/tests/api/nodes_test.py b/datajunction-server/tests/api/nodes_test.py
index bbdcead1a..7f4f3a829 100644
--- a/datajunction-server/tests/api/nodes_test.py
+++ b/datajunction-server/tests/api/nodes_test.py
@@ -5018,6 +5018,51 @@ async def test_propagate_update_downstream(
         # Node may be valid or invalid depending on whether removed columns were used
         assert data["status"] in ("valid", "invalid")
 
+    @pytest.mark.asyncio
+    async def test_propagate_update_bumps_dependent_cube_version(
+        self,
+        client_with_repairs_cube: AsyncClient,
+    ):
+        """
+        When a metric's query changes, every cube that depends on that metric must
+        get a minor version bump. AMC gates Maestro workflow pushes on the DJ cube
+        version ID, so without this bump the materialization YAML is never regenerated
+        even though the effective SQL changed (e.g. avg→sum in a pre-aggregation).
+        """
+        from tests.api.cubes_test import make_a_test_cube
+
+        await make_a_test_cube(
+            client_with_repairs_cube,
+            "default.propagation_cube",
+            with_materialization=False,
+        )
+
+        # Record the cube's version before touching any upstream metric
+        response = await client_with_repairs_cube.get(
+            "/nodes/default.propagation_cube/",
+        )
+        assert response.status_code == 200
+        version_before = response.json()["version"]
+
+        # Update an upstream metric's query — this is the avg→sum style change that
+        # should bubble up to bump the cube's version via _bump_dependent_cube_versions
+        response = await client_with_repairs_cube.patch(
+            "/nodes/default.avg_repair_price/",
+            json={"query": "SELECT SUM(price) FROM default.repair_order_details"},
+        )
+        assert response.status_code == 200
+
+        # The cube version must have incremented
+        response = await client_with_repairs_cube.get(
+            "/nodes/default.propagation_cube/",
+        )
+        assert response.status_code == 200
+        version_after = response.json()["version"]
+        assert version_after != version_before, (
+            f"Cube version should have bumped when upstream metric changed "
+            f"(was {version_before}, still {version_after})"
+        )
+
     @pytest.mark.asyncio
     async def test_update_dimension_remove_pk_column(
         self,

From 264f8a4877b159ab7f785aa243824c6c1bdc7f4a Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 01:01:20 -0700
Subject: [PATCH 02/10] Clean up structure

---
 .../internal/deployment/orchestrator.py       |  6 +++
 .../datajunction_server/internal/impact.py    | 29 +++++++++++++++
 .../datajunction_server/internal/nodes.py     | 37 ++++++++++++++++---
 3 files changed, 66 insertions(+), 6 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/deployment/orchestrator.py b/datajunction-server/datajunction_server/internal/deployment/orchestrator.py
index 766c32c83..7bd9720ac 100644
--- a/datajunction-server/datajunction_server/internal/deployment/orchestrator.py
+++ b/datajunction-server/datajunction_server/internal/deployment/orchestrator.py
@@ -1345,6 +1345,10 @@ async def _execute_deployment_plan(self, plan: DeploymentPlan) -> list:
             if r.deploy_type == DeploymentResult.Type.LINK
             and r.status != DeploymentResult.Status.SKIPPED
         }
+
+        async def _add_history(event, session):
+            session.add(event)
+
         with timer.phase("propagate impact") as p:
             downstream = await propagate_impact(
                 session=self.session,
@@ -1354,6 +1358,8 @@ async def _execute_deployment_plan(self, plan: DeploymentPlan) -> list:
                     spec.rendered_name for spec in plan.to_delete
                 ),
                 changed_link_node_names=changed_link_names,
+                current_user=self.context.current_user,
+                save_history=_add_history,
             )
             p.append(f"{len(downstream)} downstream")
 
diff --git a/datajunction-server/datajunction_server/internal/impact.py b/datajunction-server/datajunction_server/internal/impact.py
index 8181d13a7..df40bda09 100644
--- a/datajunction-server/datajunction_server/internal/impact.py
+++ b/datajunction-server/datajunction_server/internal/impact.py
@@ -64,6 +64,8 @@ async def propagate_impact(
     changed_node_names: set[str],
     deleted_node_names: frozenset[str] = frozenset(),
     changed_link_node_names: set[str] | None = None,
+    current_user=None,
+    save_history=None,
 ) -> list[DownstreamImpact]:
     """BFS downstream impact analysis with revalidation.
 
@@ -79,6 +81,10 @@ async def propagate_impact(
         deleted_node_names: Names of nodes about to be deleted (still in DB at
             call time — caller must invoke this before hard_delete_node).
         changed_link_node_names: Names of nodes whose dimension links changed.
+        current_user: If provided (with save_history), bumps the minor version of
+            every downstream cube so version-gated consumers see the change.
+        save_history: Callable(event, session) used to record the version-bump
+            history events. Required when current_user is provided.
 
     Returns:
         List of DownstreamImpact describing each affected downstream node.
@@ -112,6 +118,29 @@ async def propagate_impact(
     # Phase 3: revalidate all downstream nodes and apply status changes
     results = await _revalidate_and_apply(session, ctx, all_impacts)
 
+    # Phase 4: bump the minor version of every downstream cube so that
+    # version-gated consumers see the change. Cubes whose element list is
+    # unchanged don't get a new version from the normal diff path, but their
+    # effective SQL may have changed due to an upstream metric being updated.
+    # Only bumps cubes downstream of *changed* nodes — deleted-node downstreams
+    # become invalid and don't need a version bump.
+    if current_user and save_history and changed_node_names:
+        from datajunction_server.internal.nodes import bump_cube_versions
+
+        name_to_node = {n.name: n for n in ctx.visited_nodes_by_id.values()}
+        cube_downstreams = [
+            name_to_node[r.name]
+            for r in results
+            if r.node_type == NodeType.CUBE and r.name in name_to_node
+        ]
+        if cube_downstreams:
+            await bump_cube_versions(
+                session,
+                cube_downstreams,
+                current_user,
+                save_history,
+            )
+
     _emit_metrics(start, results)
     return results
 
diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index 0894f0043..68a7dd4a3 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1791,7 +1791,32 @@ async def _propagate_update_downstream(
     # A cube's element list doesn't change when an upstream metric's SQL
     # changes, so update_cube_node's diff logic won't fire. Create a new
     # minor revision so callers that gate on version ID see the change.
-    for cube in cube_downstreams:
+    await bump_cube_versions(
+        session,
+        cube_downstreams,
+        current_user,
+        save_history,
+        upstream_node_name=node.name,
+        upstream_node_version=node.current_version,
+    )
+
+
+async def bump_cube_versions(
+    session: AsyncSession,
+    cubes: List[Node],
+    current_user: User,
+    save_history: Callable,
+    upstream_node_name: str = "",
+    upstream_node_version: str = "",
+) -> None:
+    """
+    Create a new minor-version revision for each cube in ``cubes``.
+
+    A cube's element list is unchanged when an upstream metric's SQL changes,
+    so the normal diff-based update path produces no version bump. This function
+    forces one, ensuring any version-gated downstream consumer sees the change.
+    """
+    for cube in cubes:
         cube_node = await Node.get_cube_by_name(session, cube.name)
         if not cube_node or not cube_node.current:
             continue
@@ -1828,11 +1853,11 @@ async def _propagate_update_downstream(
                     details={
                         "version": new_cube_revision.version,
                         "upstream": {
-                            "node": node.name,
-                            "version": node.current_version,
+                            "node": upstream_node_name,
+                            "version": upstream_node_version,
                         },
-                        "reason": f"Caused by update of `{node.name}` to "
-                        f"{node.current_version}",
+                        "reason": f"Caused by update of `{upstream_node_name}` to "
+                        f"{upstream_node_version}",
                     },
                     pre={"version": current_rev.version},
                     post={"version": new_cube_revision.version},
@@ -1847,7 +1872,7 @@ async def _propagate_update_downstream(
             cube.name,
             current_rev.version,
             new_cube_revision.version,
-            node.name,
+            upstream_node_name,
         )
 
 

From a6b591a976016a77b5817c1d357bdf5eccf1d185 Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 01:02:57 -0700
Subject: [PATCH 03/10] Fix

---
 datajunction-server/datajunction_server/internal/impact.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/impact.py b/datajunction-server/datajunction_server/internal/impact.py
index df40bda09..8c16dca51 100644
--- a/datajunction-server/datajunction_server/internal/impact.py
+++ b/datajunction-server/datajunction_server/internal/impact.py
@@ -20,6 +20,7 @@
 from sqlalchemy.orm import joinedload, selectinload
 
 from datajunction_server.database.node import Node, NodeRevision, NodeRelationship
+from datajunction_server.internal.nodes import bump_cube_versions
 from datajunction_server.instrumentation.provider import get_metrics_provider
 from datajunction_server.internal.deployment.dimension_reachability import (
     DimensionReachability,
@@ -125,8 +126,6 @@ async def propagate_impact(
     # Only bumps cubes downstream of *changed* nodes — deleted-node downstreams
     # become invalid and don't need a version bump.
     if current_user and save_history and changed_node_names:
-        from datajunction_server.internal.nodes import bump_cube_versions
-
         name_to_node = {n.name: n for n in ctx.visited_nodes_by_id.values()}
         cube_downstreams = [
             name_to_node[r.name]

From 81ad51a0a0342911523ac1a6bf69cd50deb556cf Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 08:59:52 -0700
Subject: [PATCH 04/10] Fix test errors

---
 .../datajunction_server/internal/nodes.py     | 24 +++++++------------
 datajunction-server/tests/api/nodes_test.py   | 18 +++++++-------
 2 files changed, 16 insertions(+), 26 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index 68a7dd4a3..0b07ca102 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1822,23 +1822,15 @@ async def bump_cube_versions(
             continue
         current_rev = cube_node.current
 
-        create_cube = CreateCubeNode(
-            name=current_rev.name,
-            display_name=current_rev.display_name,
-            description=current_rev.description,
-            metrics=[m.name for m in current_rev.cube_metrics()],
-            dimensions=current_rev.cube_dimensions(),
-            mode=current_rev.mode,
-            filters=current_rev.cube_filters or [],
-            custom_metadata=current_rev.custom_metadata,
-        )
-
         with session.no_autoflush:
-            new_cube_revision = await create_cube_node_revision(
-                session,
-                create_cube,
-                current_user,
-            )
+            # Copy the existing revision directly so column partition settings
+            # (granularity, format, type) are preserved exactly. Using
+            # create_cube_node_revision would rebuild columns from scratch and
+            # lose those settings, breaking materialization SQL generation.
+            new_cube_revision = copy_existing_node_revision(current_rev, current_user)
+            new_cube_revision.cube_elements = current_rev.cube_elements
+            new_cube_revision.cube_filters = current_rev.cube_filters
+            new_cube_revision.materializations = []
             old_version = Version.parse(current_rev.version)
             new_cube_revision.version = str(old_version.next_minor_version())
             new_cube_revision.node = current_rev.node
diff --git a/datajunction-server/tests/api/nodes_test.py b/datajunction-server/tests/api/nodes_test.py
index 7f4f3a829..2fca2a8ee 100644
--- a/datajunction-server/tests/api/nodes_test.py
+++ b/datajunction-server/tests/api/nodes_test.py
@@ -5021,39 +5021,37 @@ async def test_propagate_update_downstream(
     @pytest.mark.asyncio
     async def test_propagate_update_bumps_dependent_cube_version(
         self,
-        client_with_repairs_cube: AsyncClient,
+        client_with_roads: AsyncClient,
     ):
         """
         When a metric's query changes, every cube that depends on that metric must
-        get a minor version bump. AMC gates Maestro workflow pushes on the DJ cube
-        version ID, so without this bump the materialization YAML is never regenerated
-        even though the effective SQL changed (e.g. avg→sum in a pre-aggregation).
+        get a minor version bump so that version-gated downstream consumers see
+        the change and regenerate any derived artifacts.
         """
         from tests.api.cubes_test import make_a_test_cube
 
         await make_a_test_cube(
-            client_with_repairs_cube,
+            client_with_roads,
             "default.propagation_cube",
             with_materialization=False,
         )
 
         # Record the cube's version before touching any upstream metric
-        response = await client_with_repairs_cube.get(
+        response = await client_with_roads.get(
             "/nodes/default.propagation_cube/",
         )
         assert response.status_code == 200
         version_before = response.json()["version"]
 
-        # Update an upstream metric's query — this is the avg→sum style change that
-        # should bubble up to bump the cube's version via _bump_dependent_cube_versions
-        response = await client_with_repairs_cube.patch(
+        # Patch an upstream metric's query to simulate a SQL change
+        response = await client_with_roads.patch(
             "/nodes/default.avg_repair_price/",
             json={"query": "SELECT SUM(price) FROM default.repair_order_details"},
         )
         assert response.status_code == 200
 
         # The cube version must have incremented
-        response = await client_with_repairs_cube.get(
+        response = await client_with_roads.get(
             "/nodes/default.propagation_cube/",
         )
         assert response.status_code == 200

From b42dcfa953e1e59998748ea4ab23aa2f495043b8 Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 09:37:11 -0700
Subject: [PATCH 05/10] Fix

---
 datajunction-server/tests/api/nodes_test.py | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/datajunction-server/tests/api/nodes_test.py b/datajunction-server/tests/api/nodes_test.py
index 2fca2a8ee..b39878c90 100644
--- a/datajunction-server/tests/api/nodes_test.py
+++ b/datajunction-server/tests/api/nodes_test.py
@@ -5028,13 +5028,17 @@ async def test_propagate_update_bumps_dependent_cube_version(
         get a minor version bump so that version-gated downstream consumers see
         the change and regenerate any derived artifacts.
         """
-        from tests.api.cubes_test import make_a_test_cube
-
-        await make_a_test_cube(
-            client_with_roads,
-            "default.propagation_cube",
-            with_materialization=False,
+        response = await client_with_roads.post(
+            "/nodes/cube/",
+            json={
+                "name": "default.propagation_cube",
+                "metrics": ["default.avg_repair_price", "default.num_repair_orders"],
+                "dimensions": ["default.hard_hat.state"],
+                "description": "Cube for version-bump propagation test",
+                "mode": "published",
+            },
         )
+        assert response.status_code < 400, response.json()
 
         # Record the cube's version before touching any upstream metric
         response = await client_with_roads.get(

From dacb549f6ac97dd6cbdbdf68572f40e2bfeaf13e Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 18:47:39 -0700
Subject: [PATCH 06/10] Fix

---
 .../datajunction_server/internal/impact.py      | 11 +++--------
 .../datajunction_server/internal/nodes.py       | 17 +++++++++++------
 2 files changed, 14 insertions(+), 14 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/impact.py b/datajunction-server/datajunction_server/internal/impact.py
index 8c16dca51..f7c1b507c 100644
--- a/datajunction-server/datajunction_server/internal/impact.py
+++ b/datajunction-server/datajunction_server/internal/impact.py
@@ -126,16 +126,11 @@ async def propagate_impact(
     # Only bumps cubes downstream of *changed* nodes — deleted-node downstreams
     # become invalid and don't need a version bump.
     if current_user and save_history and changed_node_names:
-        name_to_node = {n.name: n for n in ctx.visited_nodes_by_id.values()}
-        cube_downstreams = [
-            name_to_node[r.name]
-            for r in results
-            if r.node_type == NodeType.CUBE and r.name in name_to_node
-        ]
-        if cube_downstreams:
+        cube_names = [r.name for r in results if r.node_type == NodeType.CUBE]
+        if cube_names:
             await bump_cube_versions(
                 session,
-                cube_downstreams,
+                cube_names,
                 current_user,
                 save_history,
             )
diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index 0b07ca102..0b78c6c3a 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1791,9 +1791,11 @@ async def _propagate_update_downstream(
     # A cube's element list doesn't change when an upstream metric's SQL
     # changes, so update_cube_node's diff logic won't fire. Create a new
     # minor revision so callers that gate on version ID see the change.
+    # Extract names before the non-cube commits above expire the objects.
+    cube_names = [cube.name for cube in cube_downstreams]
     await bump_cube_versions(
         session,
-        cube_downstreams,
+        cube_names,
         current_user,
         save_history,
         upstream_node_name=node.name,
@@ -1803,21 +1805,24 @@ async def _propagate_update_downstream(
 
 async def bump_cube_versions(
     session: AsyncSession,
-    cubes: List[Node],
+    cubes: List[str],
     current_user: User,
     save_history: Callable,
     upstream_node_name: str = "",
     upstream_node_version: str = "",
 ) -> None:
     """
-    Create a new minor-version revision for each cube in ``cubes``.
+    Create a new minor-version revision for each cube name in ``cubes``.
 
     A cube's element list is unchanged when an upstream metric's SQL changes,
     so the normal diff-based update path produces no version bump. This function
     forces one, ensuring any version-gated downstream consumer sees the change.
+
+    Accepts cube names (not Node objects) so callers can extract names before
+    any session commits that would expire the Node identity-map entries.
     """
-    for cube in cubes:
-        cube_node = await Node.get_cube_by_name(session, cube.name)
+    for cube_name in cubes:
+        cube_node = await Node.get_cube_by_name(session, cube_name)
         if not cube_node or not cube_node.current:
             continue
         current_rev = cube_node.current
@@ -1861,7 +1866,7 @@ async def bump_cube_versions(
         await session.commit()
         _logger.info(
             "Bumped cube %s from %s to %s due to upstream change of %s",
-            cube.name,
+            cube_name,
             current_rev.version,
             new_cube_revision.version,
             upstream_node_name,

From 920f2f0fb77144957a57fe114cfba95cab6c9af2 Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Tue, 2 Jun 2026 23:49:13 -0700
Subject: [PATCH 07/10] Fix

---
 .../datajunction_server/internal/nodes.py     | 36 ++++++++++---------
 .../internal/nodes/background_tasks_test.py   |  3 +-
 2 files changed, 22 insertions(+), 17 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index 0b78c6c3a..4af9160a7 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1377,7 +1377,8 @@ async def update_node_with_query(
 
     background_tasks.add_task(
         propagate_update_downstream,
-        node,
+        node.name,  # type: ignore
+        node.current_version,  # type: ignore
         current_user=current_user,
         save_history=save_history,
         cache=cache,
@@ -1662,7 +1663,8 @@ async def update_cube_node(
 
 
 async def propagate_update_downstream(
-    node: Node,
+    node_name: str,
+    node_current_version: str,
     current_user: User,
     save_history: Callable,
     cache: Cache | None = None,
@@ -1674,7 +1676,8 @@ async def propagate_update_downstream(
         async with session_context() as session:
             await _propagate_update_downstream(
                 session=session,
-                node=node,
+                node_name=node_name,
+                node_current_version=node_current_version,
                 current_user=current_user,
                 save_history=save_history,
                 cache=cache,
@@ -1682,13 +1685,14 @@ async def propagate_update_downstream(
     except Exception:
         _logger.exception(
             "Error propagating update of node %s downstream",
-            node.name,
+            node_name,
         )
 
 
 async def _propagate_update_downstream(
     session: AsyncSession,
-    node: Node,
+    node_name: str,
+    node_current_version: str,
     current_user: User,
     save_history: Callable,
     cache: Cache | None = None,
@@ -1700,10 +1704,10 @@ async def _propagate_update_downstream(
     - altered column types: may invalidate downstream nodes
     - new columns: won't affect downstream nodes
     """
-    _logger.info("Propagating update of node %s downstream", node.name)
+    _logger.info("Propagating update of node %s downstream", node_name)
     all_downstreams = await get_downstream_nodes(
         session,
-        node.name,
+        node_name,
         include_deactivated=False,
         include_cubes=True,
     )
@@ -1713,7 +1717,7 @@ async def _propagate_update_downstream(
     cube_downstreams = [n for n in all_downstreams if n.type == NodeType.CUBE]
     _logger.info(
         "Node %s updated — revalidating %s downstreams, bumping %s cubes",
-        node.name,
+        node_name,
         len(non_cube_downstreams),
         len(cube_downstreams),
     )
@@ -1728,7 +1732,7 @@ async def _propagate_update_downstream(
             idx + 1,
             len(non_cube_downstreams),
             downstream.name,
-            node.name,
+            node_name,
         )
         node_validator = await revalidate_node(
             downstream.name,
@@ -1745,7 +1749,7 @@ async def _propagate_update_downstream(
                 _logger.info(
                     "Clearing upstream cache for node %s due to update of node %s (cache key: %s)",
                     downstream.name,
-                    node.name,
+                    node_name,
                     upstream_cache_key,
                 )
                 cache.delete(upstream_cache_key)
@@ -1768,11 +1772,11 @@ async def _propagate_update_downstream(
                             ),
                         },
                         "upstream": {
-                            "node": node.name,
-                            "version": node.current_version,
+                            "node": node_name,
+                            "version": node_current_version,
                         },
-                        "reason": f"Caused by update of `{node.name}` to "
-                        f"{node.current_version}",
+                        "reason": f"Caused by update of `{node_name}` to "
+                        f"{node_current_version}",
                     },
                     pre={
                         "status": previous_status,
@@ -1798,8 +1802,8 @@ async def _propagate_update_downstream(
         cube_names,
         current_user,
         save_history,
-        upstream_node_name=node.name,
-        upstream_node_version=node.current_version,
+        upstream_node_name=node_name,
+        upstream_node_version=node_current_version,
     )
 
 
diff --git a/datajunction-server/tests/internal/nodes/background_tasks_test.py b/datajunction-server/tests/internal/nodes/background_tasks_test.py
index 9dc65ee0c..376d68544 100644
--- a/datajunction-server/tests/internal/nodes/background_tasks_test.py
+++ b/datajunction-server/tests/internal/nodes/background_tasks_test.py
@@ -37,7 +37,8 @@ async def test_propagate_update_downstream_swallows_exceptions(caplog):
                 logger="datajunction_server.internal.nodes",
             ):
                 await propagate_update_downstream(
-                    node=MagicMock(name="test.node"),
+                    "test.node",
+                    "v1.0",
                     current_user=MagicMock(),
                     save_history=MagicMock(),
                 )

From 532e445ea14ab4ff97c8ed6d5a9bc5864fc65767 Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Wed, 3 Jun 2026 00:29:41 -0700
Subject: [PATCH 08/10] Fix

---
 datajunction-server/datajunction_server/internal/nodes.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index 4af9160a7..736d0b097 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1831,6 +1831,11 @@ async def bump_cube_versions(
             continue
         current_rev = cube_node.current
 
+        # Load relationships not covered by get_cube_by_name's default eager-load
+        # before entering no_autoflush. Lazy-loading inside an async session raises
+        # MissingGreenlet; explicit refresh is the correct async pattern.
+        await session.refresh(current_rev, ["parents", "dimension_links"])
+
         with session.no_autoflush:
             # Copy the existing revision directly so column partition settings
             # (granularity, format, type) are preserved exactly. Using

From efefacf36802c8514a26b0554eb1dd40bbc8f582 Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Wed, 3 Jun 2026 01:07:57 -0700
Subject: [PATCH 09/10] Fix

---
 datajunction-server/tests/api/nodes_test.py | 52 ++++++++++++++-------
 1 file changed, 36 insertions(+), 16 deletions(-)

diff --git a/datajunction-server/tests/api/nodes_test.py b/datajunction-server/tests/api/nodes_test.py
index b39878c90..12ff75ac7 100644
--- a/datajunction-server/tests/api/nodes_test.py
+++ b/datajunction-server/tests/api/nodes_test.py
@@ -5022,12 +5022,16 @@ async def test_propagate_update_downstream(
     async def test_propagate_update_bumps_dependent_cube_version(
         self,
         client_with_roads: AsyncClient,
+        session: AsyncSession,
     ):
         """
         When a metric's query changes, every cube that depends on that metric must
         get a minor version bump so that version-gated downstream consumers see
         the change and regenerate any derived artifacts.
         """
+        from sqlalchemy.orm import joinedload
+        from datajunction_server.internal.nodes import _propagate_update_downstream
+
         response = await client_with_roads.post(
             "/nodes/cube/",
             json={
@@ -5040,29 +5044,45 @@ async def test_propagate_update_bumps_dependent_cube_version(
         )
         assert response.status_code < 400, response.json()
 
-        # Record the cube's version before touching any upstream metric
-        response = await client_with_roads.get(
-            "/nodes/default.propagation_cube/",
-        )
-        assert response.status_code == 200
-        version_before = response.json()["version"]
-
-        # Patch an upstream metric's query to simulate a SQL change
         response = await client_with_roads.patch(
             "/nodes/default.avg_repair_price/",
             json={"query": "SELECT SUM(price) FROM default.repair_order_details"},
         )
         assert response.status_code == 200
 
-        # The cube version must have incremented
-        response = await client_with_roads.get(
-            "/nodes/default.propagation_cube/",
-        )
-        assert response.status_code == 200
-        version_after = response.json()["version"]
-        assert version_after != version_before, (
+        metric = await Node.get_by_name(
+            session,
+            "default.avg_repair_price",
+            options=[joinedload(Node.current)],
+        )
+        cube = await Node.get_by_name(
+            session,
+            "default.propagation_cube",
+            options=[joinedload(Node.current)],
+        )
+        assert metric is not None
+        assert cube is not None
+        version_before = cube.current_version
+        user = await User.get_by_username(session, "dj")
+        assert user is not None
+
+        async def noop_save_history(event, session):
+            session.add(event)
+
+        # Call propagation directly with the test session so the newly created
+        # cube (not yet committed to the DB) is visible.
+        await _propagate_update_downstream(
+            session=session,
+            node_name=metric.name,
+            node_current_version=metric.current_version,
+            current_user=user,
+            save_history=noop_save_history,
+        )
+
+        await session.refresh(cube)
+        assert cube.current_version != version_before, (
             f"Cube version should have bumped when upstream metric changed "
-            f"(was {version_before}, still {version_after})"
+            f"(was {version_before}, still {cube.current_version})"
         )
 
     @pytest.mark.asyncio

From b3cd6a3a58b2d9f62a0ededb2f806a1e8e2a1159 Mon Sep 17 00:00:00 2001
From: Yian Shang <yian.shang@gmail.com>
Date: Wed, 3 Jun 2026 08:50:56 -0700
Subject: [PATCH 10/10] Fix

---
 datajunction-server/datajunction_server/internal/nodes.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py
index 736d0b097..d43f55794 100644
--- a/datajunction-server/datajunction_server/internal/nodes.py
+++ b/datajunction-server/datajunction_server/internal/nodes.py
@@ -1715,6 +1715,9 @@ async def _propagate_update_downstream(
         [n for n in all_downstreams if n.type != NodeType.CUBE],
     )
     cube_downstreams = [n for n in all_downstreams if n.type == NodeType.CUBE]
+    # Extract names now — commits in the loop below expire the Node objects,
+    # and accessing .name on an expired async-session object raises MissingGreenlet.
+    cube_names = [cube.name for cube in cube_downstreams]
     _logger.info(
         "Node %s updated — revalidating %s downstreams, bumping %s cubes",
         node_name,
@@ -1795,8 +1798,6 @@ async def _propagate_update_downstream(
     # A cube's element list doesn't change when an upstream metric's SQL
     # changes, so update_cube_node's diff logic won't fire. Create a new
     # minor revision so callers that gate on version ID see the change.
-    # Extract names before the non-cube commits above expire the objects.
-    cube_names = [cube.name for cube in cube_downstreams]
     await bump_cube_versions(
         session,
         cube_names,