Run stream of deferred queries sequentially

core/src/main/scala/caliban/execution/Executor.scala

@@ -231,7 +231,7 @@ object Executor {
             ZStream.succeed(resp) ++ makeDeferStream(more, cache)
         })
 
-      ZStream.mergeAllUnbounded()(defers.map(run): _*)
+      ZStream.mergeAll(1)(defers.map(run): _*)
     }
 
     def runIncrementalQuery(

core/src/test/scala/caliban/execution/DeferredExecutionSpec.scala

@@ -5,8 +5,8 @@ import caliban.ResponseValue.StreamValue
 import caliban.TestUtils.{ characters, Character }
 import caliban.{ CalibanError, GraphQLResponse, ResponseValue, Value }
 import zio.test.Assertion.hasSameElements
-import zio.test.{ assert, assertTrue, TestAspect, ZIOSpecDefault }
-import zio.{ Chunk, UIO, ZIO, ZLayer }
+import zio.test.{ assert, assertTrue, TestAspect, TestClock, ZIOSpecDefault }
+import zio._
 
 trait CharacterService {
   def characterBy(pred: Character => Boolean): UIO[List[Character]]
@@ -188,7 +188,32 @@ object DeferredExecutionSpec extends ZIOSpecDefault {
           """{"hasNext":false}"""
         )
       )
-    }
+    },
+    test("deferred fields backed by a datasource") {
+      import TestDatasourceDeferredSchema._
+      val query = gqldoc("""
+        {
+          foo {
+            bar { ... @defer { value } }
+          }
+        }""")
+
+      for {
+        resp <- interpreter.flatMap(_.execute(query))
+        rest <- runIncrementalResponses(resp)
+        head  = rest.head.toString
+        mid   = rest.tail.init.toList.map(_.toString)
+        last  = rest.last.toString
+      } yield assertTrue(
+        head == """{"foo":{"bar":[{},{},{}]}}""",
+        mid.sorted == List(
+          """{"incremental":[{"data":{"value":"value"},"path":["foo","bar",0]}],"hasNext":true}""",
+          """{"incremental":[{"data":{"value":"value"},"path":["foo","bar",1]}],"hasNext":true}""",
+          """{"incremental":[{"data":{"value":"value"},"path":["foo","bar",2]}],"hasNext":true}"""
+        ),
+        last == """{"hasNext":false}"""
+      )
+    } @@ TestAspect.nonFlaky(50)
   ).provide(CharacterService.test)
 
   def runIncrementalResponses(response: GraphQLResponse[CalibanError]) =

core/src/test/scala/caliban/execution/TestDeferredSchema.scala

@@ -6,6 +6,7 @@ import caliban.{ graphQL, RootResolver }
 import caliban.schema.Annotations.GQLName
 import caliban.schema.{ GenericSchema, Schema }
 import caliban.wrappers.DeferSupport
+import zio.query.{ DataSource, Request, UQuery, ZQuery }
 import zio.{ UIO, URIO, ZIO }
 
 object TestDeferredSchema extends GenericSchema[CharacterService] {
@@ -77,3 +78,22 @@ object TestDeferredSchema extends GenericSchema[CharacterService] {
 
   val interpreter = (graphQL(resolver) @@ DeferSupport.defer).interpreter
 }
+
+object TestDatasourceDeferredSchema extends GenericSchema[Any] {
+  import auto._
+
+  case class Bar(value: UQuery[String])
+  case class Foo(bar: UIO[List[Bar]])
+  case class Query(foo: UIO[Foo])
+
+  case class Req(i: Int) extends Request[Nothing, String]
+  private val ds = DataSource.fromFunctionZIO("ValuesDS")((_: Req) => ZIO.succeed("value"))
+
+  private def makeBar(i: Int): Bar = Bar(ZQuery.fromRequest(Req(i))(ds))
+
+  private val resolver = RootResolver(Query(ZIO.succeed {
+    Foo(bar = ZIO.succeed(List(makeBar(1), makeBar(3), makeBar(3))))
+  }))
+
+  val interpreter = (graphQL(resolver) @@ DeferSupport.defer).interpreter
+}