Avoid hasNext=true on the last incremental payload

lib/src/main/java/graphql/nadel/engine/NadelIncrementalResultAccumulator.kt

@@ -40,7 +40,7 @@ import graphql.normalized.incremental.NormalizedDeferredExecution
  * the user.
  */
 class NadelIncrementalResultAccumulator(
-    private val operation: ExecutableNormalizedOperation,
+    operation: ExecutableNormalizedOperation,
 ) {
     data class DeferAccumulatorKey(
         val incrementalPayloadPath: List<Any>,
@@ -115,6 +115,14 @@ class NadelIncrementalResultAccumulator(
             }
 
         if (readyAccumulators.isEmpty()) {
+            if(!hasNext) {
+                // We have to return hasNext=false to indicate to clients that there's no more data coming.
+                // Note: the spec allows an empty payload which only contains hastNext=false to be returned.
+                return DelayedIncrementalPartialResultImpl.newIncrementalExecutionResult()
+                    .incrementalItems(emptyList())
+                    .hasNext(false)
+                    .build()
+            }
             return null
         }
 

lib/src/main/java/graphql/nadel/engine/NadelIncrementalResultSupport.kt

@@ -2,15 +2,13 @@ package graphql.nadel.engine
 
 import graphql.incremental.DelayedIncrementalPartialResult
 import graphql.nadel.engine.NadelIncrementalResultSupport.OutstandingJobCounter.OutstandingJobHandle
-import graphql.nadel.engine.util.copy
 import graphql.nadel.util.getLogger
 import graphql.normalized.ExecutableNormalizedOperation
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
@@ -145,21 +143,6 @@ class NadelIncrementalResultSupport internal constructor(
         initialCompletionLock.complete(Unit)
     }
 
-    fun close() {
-        coroutineScope.cancel()
-    }
-
-    private fun quickCopy(
-        subject: DelayedIncrementalPartialResult,
-        hasNext: Boolean,
-    ): DelayedIncrementalPartialResult {
-        return if (subject.hasNext() == hasNext) {
-            subject
-        } else {
-            subject.copy(hasNext = hasNext)
-        }
-    }
-
     /**
      * Launches a job and increments the outstanding job handle.
      *

lib/src/test/kotlin/graphql/nadel/engine/NadelIncrementalResultAccumulatorTest.kt

@@ -174,12 +174,42 @@ class NadelIncrementalResultAccumulatorTest {
                 )
 
                 // Then
-                val result = accumulator.getIncrementalPartialResult(false)
+                val result = accumulator.getIncrementalPartialResult(true)
                 assertTrue(result == null)
             }
         }
     }
 
+    @Test
+    fun `yields empty payload if hasNext=false and accumulator is empty`() {
+        val accumulator = makeAccumulator(
+            query = """
+                        query {
+                          issue(id: "1") {
+                            ... @defer {
+                              id
+                              assignee {
+                                name
+                              }
+                            }
+                          }
+                        }
+                    """.trimIndent(),
+        )
+
+        // When
+        accumulator.accumulate(
+            DelayedIncrementalPartialResultImpl.newIncrementalExecutionResult()
+                .incrementalItems(emptyList())
+                .build(),
+        )
+
+        // Then
+        val result = accumulator.getIncrementalPartialResult(false)
+        assertTrue(result!!.incremental!!.isEmpty())
+        assertFalse(result.hasNext())
+    }
+
     @Test
     fun `yields list elements that are complete`() {
         val accumulator = makeAccumulator(

test/src/test/kotlin/graphql/nadel/tests/next/NadelIntegrationTest.kt

@@ -379,7 +379,9 @@ abstract class NadelIntegrationTest(
             // Note: there exists a IncrementalExecutionResult.getIncremental but that is part of the initial result
             assertTrue(result is IncrementalExecutionResult)
 
-            // Fuck why delayed & incremental?? Shouldn't incremental == delayed? Why is there an optional synchronous incremental??
+            // Note: the spec allows for a list of "incremental" results which is sent as part of the initial
+            // result (so not delivered in a delayed fashion). This var represents the incremental results that were
+            // sent in a delayed fashion.
             val actualDelayedResponses = incrementalResults!!
 
             // Should only have one element that says hasNext=false, and it should be the last one