Fix block reference load in async flow (#15487)

Co-authored-by: Ladislav Gál <[email protected]>

src/prefect/flows.py

@@ -535,7 +535,7 @@ def validate_parameters(self, parameters: Dict[str, Any]) -> Dict[str, Any]:
 
         def resolve_block_reference(data: Any) -> Any:
             if isinstance(data, dict) and "$ref" in data:
-                return Block.load_from_ref(data["$ref"])
+                return Block.load_from_ref(data["$ref"], _sync=True)
             return data
 
         try:

tests/blocks/test_block_reference.py

@@ -1,3 +1,4 @@
+import asyncio
 import warnings
 from typing import Type
 from uuid import UUID, uuid4
@@ -189,3 +190,26 @@ def flow_with_block_param_in_basemodel(param: ParamModel):
             )
             == param_block.a
         )
+
+    def test_async_flow_with_block_params(self, ParamBlock):
+        ref_block = ParamBlock(a=10, b="foo")
+        ref_block.save("param-block")
+
+        @flow
+        async def flow_with_block_param(block: ParamBlock) -> int:
+            return block.a
+
+        assert (
+            asyncio.run(
+                flow_with_block_param({"$ref": str(ref_block._block_document_id)})
+            )
+            == ref_block.a
+        )
+        assert (
+            asyncio.run(
+                flow_with_block_param(
+                    {"$ref": {"block_document_id": str(ref_block._block_document_id)}}
+                )
+            )
+            == ref_block.a
+        )