test: cleanup defer/stream source that never publishes a source :) (#…

…2083)

* test: cleanup defer/stream source that never publishes a source :)

* Bump executor

* Go

* Bump fetch

Co-authored-by: Arda TANRIKULU <ardatanrikulu@gmail.com>

examples/bun/package.json

@@ -12,6 +12,6 @@
     "graphql": "16.6.0"
   },
   "devDependencies": {
-    "@whatwg-node/fetch": "0.5.1"
+    "@whatwg-node/fetch": "0.5.3"
   }
 }

examples/error-handling/package.json

@@ -8,7 +8,7 @@
   },
   "dependencies": {
     "graphql-yoga": "3.0.0-next.11",
-    "@whatwg-node/fetch": "0.5.1",
+    "@whatwg-node/fetch": "0.5.3",
     "graphql": "^16.1.0"
   },
   "devDependencies": {

packages/graphql-yoga/package.json

@@ -52,12 +52,12 @@
     "@envelop/core": "3.0.3",
     "@envelop/parser-cache": "5.0.3",
     "@envelop/validation-cache": "5.0.3",
-    "@graphql-tools/executor": "0.0.7",
+    "@graphql-tools/executor": "0.0.8",
     "@graphql-tools/schema": "^9.0.0",
     "@graphql-tools/utils": "^9.0.1",
     "@graphql-yoga/subscription": "^3.0.0-next.0",
-    "@whatwg-node/fetch": "0.5.1",
-    "@whatwg-node/server": "0.4.14",
+    "@whatwg-node/fetch": "0.5.3",
+    "@whatwg-node/server": "0.4.16",
     "dset": "^3.1.1",
     "tslib": "^2.3.1"
   },

packages/plugins/apq/package.json

@@ -40,7 +40,7 @@
     "access": "public"
   },
   "dependencies": {
-    "@whatwg-node/fetch": "0.5.1",
+    "@whatwg-node/fetch": "0.5.3",
     "tiny-lru": "^10.0.0"
   },
   "devDependencies": {

packages/plugins/defer-stream/__integration-tests__/defer-stream.spec.ts

@@ -1,8 +1,9 @@
 import { createSchema, createYoga } from 'graphql-yoga'
 import { createServer } from 'node:http'
-import { fetch } from '@whatwg-node/fetch'
 import { createPushPullAsyncIterable } from '../__tests__/push-pull-async-iterable.js'
 import { useDeferStream } from '@graphql-yoga/plugin-defer-stream'
+import { fetch, AbortController } from '@whatwg-node/fetch'
+import { get, IncomingMessage } from 'http'
 
 it('correctly deals with the source upon aborted requests', async () => {
   const { source, push, terminate } = createPushPullAsyncIterable<string>()
@@ -91,3 +92,108 @@ it('correctly deals with the source upon aborted requests', async () => {
     })
   }
 })
+
+it('memory/cleanup leak by source that never publishes a value', async () => {
+  let sourceGotCleanedUp = false
+  let i = 1
+  let interval: any
+  const controller = new AbortController()
+
+  const noop = new Promise<{ done: true; value: undefined }>(() => undefined)
+  const source = {
+    [Symbol.asyncIterator]() {
+      return this
+    },
+    next() {
+      interval = setInterval(() => {
+        i++
+        if (i === 3) {
+          controller.abort()
+        }
+      }, 10)
+      return noop
+    },
+    return() {
+      clearInterval(interval)
+      sourceGotCleanedUp = true
+      return Promise.resolve({ done: true, value: undefined })
+    },
+  }
+
+  const yoga = createYoga({
+    schema: createSchema({
+      typeDefs: /* GraphQL */ `
+        type Query {
+          hi: [String]
+        }
+      `,
+      resolvers: {
+        Query: {
+          hi: () => source,
+        },
+      },
+    }),
+    plugins: [useDeferStream()],
+  })
+
+  const server = createServer(yoga)
+  try {
+    await new Promise<void>((resolve) => {
+      server.listen(() => {
+        resolve()
+      })
+    })
+
+    const port = (server.address() as any)?.port ?? null
+    if (port === null) {
+      throw new Error('Missing port...')
+    }
+
+    const response = await new Promise<IncomingMessage>((resolve) => {
+      const request = get(
+        `http://localhost:${port}/graphql?query={hi @stream}`,
+        {
+          headers: {
+            accept: 'multipart/mixed',
+          },
+        },
+        (res) => resolve(res),
+      )
+      controller.signal.addEventListener(
+        'abort',
+        () => {
+          request.destroy()
+        },
+        { once: true },
+      )
+    })
+
+    try {
+      for await (const chunk of response) {
+        const chunkStr = Buffer.from(chunk).toString('utf-8')
+        expect(chunkStr).toMatchInlineSnapshot(`
+          "---
+          Content-Type: application/json; charset=utf-8
+          Content-Length: 33
+
+          {"data":{"hi":[]},"hasNext":true}
+          ---"
+        `)
+      }
+    } catch (err: any) {
+      expect(err.message).toContain('aborted')
+    }
+
+    // Wait a bit - just to make sure the time is cleaned up for sure...
+    await new Promise((res) => setTimeout(res, 50))
+
+    expect(i).toEqual(3)
+    expect(sourceGotCleanedUp).toBe(true)
+  } finally {
+    await new Promise<void>((res) => {
+      server.close(() => {
+        res()
+      })
+    })
+  }
+})

packages/plugins/defer-stream/__tests__/defer-stream.spec.ts

@@ -91,12 +91,10 @@ describe('Defer/Stream', () => {
     expect(response.headers.get('content-type')).toBe(
       'multipart/mixed; boundary="-"',
     )
-    const chunks: string[] = []
-    for await (const chunk of response.body!) {
-      chunks.push(chunk.toString())
-    }
 
-    expect(chunks.join('')).toMatchInlineSnapshot(`
+    const finalText = await response.text()
+
+    expect(finalText).toMatchInlineSnapshot(`
       "---
       Content-Type: application/json; charset=utf-8
       Content-Length: 26
@@ -131,12 +129,10 @@ describe('Defer/Stream', () => {
     expect(response.headers.get('content-type')).toBe(
       'multipart/mixed; boundary="-"',
     )
-    const chunks: string[] = []
-    for await (const chunk of response.body!) {
-      chunks.push(chunk.toString())
-    }
 
-    expect(chunks.join('')).toMatchInlineSnapshot(`
+    const finalText = await response.text()
+
+    expect(finalText).toMatchInlineSnapshot(`
       "---
       Content-Type: application/json; charset=utf-8
       Content-Length: 44

packages/plugins/defer-stream/package.json

@@ -44,7 +44,7 @@
     "graphql-yoga": "^3.0.0-next.11"
   },
   "devDependencies": {
-    "@whatwg-node/fetch": "0.5.1"
+    "@whatwg-node/fetch": "0.5.3"
   },
   "type": "module",
   "dependencies": {