-
Notifications
You must be signed in to change notification settings - Fork 349
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[v1] fix(runtime): cleanup transport executors per schema change (#7215)
* fix(runtime): cleanup transport executors per schema change * Fix * Dispose transports on global shutdow * Subscription cleanup * Handle asynciterable results in execute as well * Fix duplicate call
- Loading branch information
Showing
8 changed files
with
230 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| --- | ||
| '@graphql-mesh/fusion-runtime': patch | ||
| '@graphql-mesh/serve-runtime': patch | ||
| --- | ||
|
|
||
| Cleanup created transport executors per schema change | ||
| Previously they were cleaned up only on server close, which could lead to memory leaks in case of schema changes. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,43 +1,103 @@ | ||
| import { buildSchema } from 'graphql'; | ||
| import { buildSchema, GraphQLSchema, parse } from 'graphql'; | ||
| import { createSchema } from 'graphql-yoga'; | ||
| import { composeSubgraphs, getUnifiedGraphGracefully } from '@graphql-mesh/fusion-composition'; | ||
| import { createDefaultExecutor, type DisposableExecutor } from '@graphql-mesh/transport-common'; | ||
| import { normalizedExecutor } from '@graphql-tools/executor'; | ||
| import { isAsyncIterable } from '@graphql-tools/utils'; | ||
| import { UnifiedGraphManager } from '../src/unifiedGraphManager'; | ||
|
|
||
| describe('Polling', () => { | ||
| it('polls the schema in a certain interval', async () => { | ||
| jest.useFakeTimers(); | ||
| const pollingInterval = 35_000; | ||
| const unifiedGraphFetcher = () => | ||
| getUnifiedGraphGracefully([ | ||
| { | ||
| name: 'Test', | ||
| schema: buildSchema(/* GraphQL */ ` | ||
| let schema: GraphQLSchema; | ||
| const unifiedGraphFetcher = () => { | ||
| const time = new Date().toISOString(); | ||
| schema = createSchema({ | ||
| typeDefs: /* GraphQL */ ` | ||
| """ | ||
| Fetched on ${new Date().toISOString()} | ||
| Fetched on ${time} | ||
| """ | ||
| type Query { | ||
| test: String | ||
| time: String | ||
| } | ||
| `), | ||
| `, | ||
| resolvers: { | ||
| Query: { | ||
| time() { | ||
| return time; | ||
| }, | ||
| }, | ||
| }, | ||
| }); | ||
| return getUnifiedGraphGracefully([ | ||
| { | ||
| name: 'Test', | ||
| schema, | ||
| }, | ||
| ]); | ||
| await using manager = new UnifiedGraphManager({ | ||
| }; | ||
| const disposeFn = jest.fn(); | ||
| const manager = new UnifiedGraphManager({ | ||
| getUnifiedGraph: unifiedGraphFetcher, | ||
| polling: pollingInterval, | ||
| transports() { | ||
| return { | ||
| getSubgraphExecutor() { | ||
| const executor: DisposableExecutor = createDefaultExecutor(schema); | ||
| executor[Symbol.asyncDispose] = disposeFn; | ||
| return executor; | ||
| }, | ||
| }; | ||
| }, | ||
| }); | ||
| async function getFetchedTime() { | ||
| async function getFetchedTimeOnComment() { | ||
| const schema = await manager.getUnifiedGraph(); | ||
| const queryType = schema.getQueryType(); | ||
| const lastFetchedDateStr = queryType.description.match(/Fetched on (.*)/)[1]; | ||
| const lastFetchedDate = new Date(lastFetchedDateStr); | ||
| return lastFetchedDate; | ||
| } | ||
| const firstDate = await getFetchedTime(); | ||
| async function getFetchedTimeFromResolvers() { | ||
| const schema = await manager.getUnifiedGraph(); | ||
| const result = await normalizedExecutor({ | ||
| schema, | ||
| document: parse(/* GraphQL */ ` | ||
| query { | ||
| time | ||
| } | ||
| `), | ||
| }); | ||
| if (isAsyncIterable(result)) { | ||
| throw new Error('Unexpected async iterable'); | ||
| } | ||
| return new Date(result.data.time); | ||
| } | ||
| async function compareTimes() { | ||
| const timeFromComment = await getFetchedTimeOnComment(); | ||
| const timeFromResolvers = await getFetchedTimeFromResolvers(); | ||
| expect(timeFromComment).toEqual(timeFromResolvers); | ||
| } | ||
| await compareTimes(); | ||
| const firstDate = await getFetchedTimeOnComment(); | ||
| jest.advanceTimersByTime(pollingInterval); | ||
| const secondDate = await getFetchedTime(); | ||
| expect(secondDate.getTime() - firstDate.getTime()).toBeGreaterThanOrEqual(pollingInterval); | ||
| await compareTimes(); | ||
| const secondDate = await getFetchedTimeOnComment(); | ||
| const diffBetweenFirstAndSecond = secondDate.getTime() - firstDate.getTime(); | ||
| expect(diffBetweenFirstAndSecond).toBeGreaterThanOrEqual(pollingInterval); | ||
| jest.advanceTimersByTime(pollingInterval); | ||
| const thirdDate = await getFetchedTime(); | ||
| expect(thirdDate.getTime() - secondDate.getTime()).toBeGreaterThanOrEqual(pollingInterval); | ||
| expect(thirdDate.getTime() - firstDate.getTime()).toBeGreaterThanOrEqual(pollingInterval * 2); | ||
| await compareTimes(); | ||
| const thirdDate = await getFetchedTimeOnComment(); | ||
| const diffBetweenSecondAndThird = thirdDate.getTime() - secondDate.getTime(); | ||
| expect(diffBetweenSecondAndThird).toBeGreaterThanOrEqual(pollingInterval); | ||
| const diffBetweenFirstAndThird = thirdDate.getTime() - firstDate.getTime(); | ||
| expect(diffBetweenFirstAndThird).toBeGreaterThanOrEqual(pollingInterval * 2); | ||
|
|
||
| // Check if transport executor is disposed per schema change | ||
| expect(disposeFn).toHaveBeenCalledTimes(2); | ||
|
|
||
| await manager[Symbol.asyncDispose](); | ||
| // Check if transport executor is disposed on global shutdown | ||
| expect(disposeFn).toHaveBeenCalledTimes(3); | ||
| }); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.