feat: improve memory management & request deduplication

execution/engine/testdata/complex_nesting_query_with_art.json

@@ -170,7 +170,7 @@
                 "duration_since_start_pretty": "1ns",
                 "duration_load_nanoseconds": 1,
                 "duration_load_pretty": "1ns",
-                "single_flight_used": false,
+                "single_flight_used": true,
                 "single_flight_shared_response": false,
                 "load_skipped": false,
                 "load_stats": {
@@ -310,7 +310,7 @@
                     "duration_since_start_pretty": "1ns",
                     "duration_load_nanoseconds": 1,
                     "duration_load_pretty": "1ns",
-                    "single_flight_used": false,
+                    "single_flight_used": true,
                     "single_flight_shared_response": false,
                     "load_skipped": false,
                     "load_stats": {
@@ -496,7 +496,7 @@
                     "duration_since_start_pretty": "1ns",
                     "duration_load_nanoseconds": 1,
                     "duration_load_pretty": "1ns",
-                    "single_flight_used": false,
+                    "single_flight_used": true,
                     "single_flight_shared_response": false,
                     "load_skipped": false,
                     "load_stats": {

go.work.sum

@@ -246,6 +246,8 @@ github.com/twmb/franz-go/pkg/kmsg v1.7.0/go.mod h1:se9Mjdt0Nwzc9lnjJ0HyDtLyBnaBD
 github.com/vbatts/tar-split v0.12.1 h1:CqKoORW7BUWBe7UL/iqTVvkTBOF8UvOMKOIZykxnnbo=
 github.com/vbatts/tar-split v0.12.1/go.mod h1:eF6B6i6ftWQcDqEn3/iGFRFRo8cBIMSJVOpnNdfTMFA=
 github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/fJgbpc=
+github.com/wundergraph/go-arena v0.0.0-20251008210416-55cb97e6f68f h1:5snewyMaIpajTu4wj22L/DgrGimICqXtUVjkZInBH3Y=
+github.com/wundergraph/go-arena v0.0.0-20251008210416-55cb97e6f68f/go.mod h1:ROOysEHWJjLQ8FSfNxZCziagb7Qw2nXY3/vgKRh7eWw=
 github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=

v2/go.mod

@@ -28,7 +28,8 @@ require (
 	github.com/tidwall/gjson v1.17.0
 	github.com/tidwall/sjson v1.2.5
 	github.com/vektah/gqlparser/v2 v2.5.30
-	github.com/wundergraph/astjson v0.0.0-20250106123708-be463c97e083
+	github.com/wundergraph/astjson v1.0.0
+	github.com/wundergraph/go-arena v1.0.0
 	go.uber.org/atomic v1.11.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.26.0

v2/go.sum

@@ -134,8 +134,10 @@ github.com/urfave/cli/v2 v2.27.7 h1:bH59vdhbjLv3LAvIu6gd0usJHgoTTPhCFib8qqOwXYU=
 github.com/urfave/cli/v2 v2.27.7/go.mod h1:CyNAG/xg+iAOg0N4MPGZqVmv2rCoP267496AOXUZjA4=
 github.com/vektah/gqlparser/v2 v2.5.30 h1:EqLwGAFLIzt1wpx1IPpY67DwUujF1OfzgEyDsLrN6kE=
 github.com/vektah/gqlparser/v2 v2.5.30/go.mod h1:D1/VCZtV3LPnQrcPBeR/q5jkSQIPti0uYCP/RI0gIeo=
-github.com/wundergraph/astjson v0.0.0-20250106123708-be463c97e083 h1:8/D7f8gKxTBjW+SZK4mhxTTBVpxcqeBgWF1Rfmltbfk=
-github.com/wundergraph/astjson v0.0.0-20250106123708-be463c97e083/go.mod h1:eOTL6acwctsN4F3b7YE+eE2t8zcJ/doLm9sZzsxxxrE=
+github.com/wundergraph/astjson v1.0.0 h1:rETLJuQkMWWW03HCF6WBttEBOu8gi5vznj5KEUPVV2Q=
+github.com/wundergraph/astjson v1.0.0/go.mod h1:h12D/dxxnedtLzsKyBLK7/Oe4TAoGpRVC9nDpDrZSWw=
+github.com/wundergraph/go-arena v1.0.0 h1:RVYWpDkJ1/6851BRHYehBeEcTLKmZygYIZsvBorcOjw=
+github.com/wundergraph/go-arena v1.0.0/go.mod h1:ROOysEHWJjLQ8FSfNxZCziagb7Qw2nXY3/vgKRh7eWw=
 github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 h1:FnBeRrxr7OU4VvAzt5X7s6266i6cSVkkFPS0TuXWbIg=
 github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=

v2/pkg/astnormalization/uploads/upload_finder.go

@@ -74,7 +74,7 @@ func (v *UploadFinder) FindUploads(operation, definition *ast.Document, variable
 		variables = []byte("{}")
 	}
 
-	v.variables, err = astjson.ParseBytesWithoutCache(variables)
+	v.variables, err = astjson.ParseBytes(variables)
 	if err != nil {
 		return nil, err
 	}

v2/pkg/astvisitor/visitor.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/wundergraph/go-arena"
+
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/ast"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/lexer/literal"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/operationreport"
@@ -95,6 +97,8 @@ type Walker struct {
 	deferred        []func()
 
 	OnExternalError func(err *operationreport.ExternalError)
+
+	arena arena.Arena
 }
 
 func NewWalkerWithID(ancestorSize int, id string) Walker {
@@ -140,6 +144,9 @@ func WalkerFromPool() *Walker {
 }
 
 func (w *Walker) Release() {
+	if w.arena != nil {
+		w.arena.Reset()
+	}
 	w.ResetVisitors()
 	w.Report = nil
 	w.document = nil
@@ -1391,6 +1398,11 @@ func (w *Walker) Walk(document, definition *ast.Document, report *operationrepor
 	} else {
 		w.Report = report
 	}
+	if w.arena == nil {
+		w.arena = arena.NewMonotonicArena(arena.WithMinBufferSize(64))
+	} else {
+		w.arena.Reset()
+	}
 	w.Ancestors = w.Ancestors[:0]
 	w.Path = w.Path[:0]
 	w.TypeDefinitions = w.TypeDefinitions[:0]
@@ -1843,8 +1855,7 @@ func (w *Walker) walkSelectionSet(ref int, skipFor SkipVisitors) {
 
 RefsChanged:
 	for {
-		refs := make([]int, 0, len(w.document.SelectionSets[ref].SelectionRefs))
-		refs = append(refs, w.document.SelectionSets[ref].SelectionRefs...)
+		refs := arena.SliceAppend(w.arena, nil, w.document.SelectionSets[ref].SelectionRefs...)
 
 		for i, j := range refs {
 

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go

@@ -14,7 +14,6 @@ import (
 	"unicode"
 
 	"github.com/buger/jsonparser"
-	"github.com/cespare/xxhash/v2"
 	"github.com/jensneuse/abstractlogger"
 	"github.com/pkg/errors"
 	"github.com/tidwall/sjson"
@@ -1907,20 +1906,19 @@ func (s *Source) replaceEmptyObject(variables []byte) ([]byte, bool) {
 	return variables, false
 }
 
-func (s *Source) LoadWithFiles(ctx context.Context, input []byte, files []*httpclient.FileUpload, out *bytes.Buffer) (err error) {
+func (s *Source) LoadWithFiles(ctx context.Context, headers http.Header, input []byte, files []*httpclient.FileUpload) (data []byte, err error) {
 	input = s.compactAndUnNullVariables(input)
-	return httpclient.DoMultipartForm(s.httpClient, ctx, input, files, out)
+	return httpclient.DoMultipartForm(s.httpClient, ctx, headers, input, files)
 }
 
-func (s *Source) Load(ctx context.Context, input []byte, out *bytes.Buffer) (err error) {
+func (s *Source) Load(ctx context.Context, headers http.Header, input []byte) (data []byte, err error) {
 	input = s.compactAndUnNullVariables(input)
-	return httpclient.Do(s.httpClient, ctx, input, out)
+	return httpclient.Do(s.httpClient, ctx, headers, input)
 }
 
 type GraphQLSubscriptionClient interface {
 	// Subscribe to the origin source. The implementation must not block the calling goroutine.
 	Subscribe(ctx *resolve.Context, options GraphQLSubscriptionOptions, updater resolve.SubscriptionUpdater) error
-	UniqueRequestID(ctx *resolve.Context, options GraphQLSubscriptionOptions, hash *xxhash.Digest) (err error)
 	SubscribeAsync(ctx *resolve.Context, id uint64, options GraphQLSubscriptionOptions, updater resolve.SubscriptionUpdater) error
 	Unsubscribe(id uint64)
 }
@@ -1956,12 +1954,13 @@ type SubscriptionSource struct {
 	client GraphQLSubscriptionClient
 }
 
-func (s *SubscriptionSource) AsyncStart(ctx *resolve.Context, id uint64, input []byte, updater resolve.SubscriptionUpdater) error {
+func (s *SubscriptionSource) AsyncStart(ctx *resolve.Context, id uint64, headers http.Header, input []byte, updater resolve.SubscriptionUpdater) error {
 	var options GraphQLSubscriptionOptions
 	err := json.Unmarshal(input, &options)
 	if err != nil {
 		return err
 	}
+	options.Header = headers
 	if options.Body.Query == "" {
 		return resolve.ErrUnableToResolve
 	}
@@ -1975,12 +1974,13 @@ func (s *SubscriptionSource) AsyncStop(id uint64) {
 }
 
 // Start the subscription. The updater is called on new events. Start needs to be called in a separate goroutine.
-func (s *SubscriptionSource) Start(ctx *resolve.Context, input []byte, updater resolve.SubscriptionUpdater) error {
+func (s *SubscriptionSource) Start(ctx *resolve.Context, headers http.Header, input []byte, updater resolve.SubscriptionUpdater) error {
 	var options GraphQLSubscriptionOptions
 	err := json.Unmarshal(input, &options)
 	if err != nil {
 		return err
 	}
+	options.Header = headers
 	if options.Body.Query == "" {
 		return resolve.ErrUnableToResolve
 	}
@@ -1990,16 +1990,3 @@ func (s *SubscriptionSource) Start(ctx *resolve.Context, input []byte, updater r
 var (
 	dataSouceName = []byte("graphql")
 )
-
-func (s *SubscriptionSource) UniqueRequestID(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) (err error) {
-	_, err = xxh.Write(dataSouceName)
-	if err != nil {
-		return err
-	}
-	var options GraphQLSubscriptionOptions
-	err = json.Unmarshal(input, &options)
-	if err != nil {
-		return err
-	}
-	return s.client.UniqueRequestID(ctx, options, xxh)
-}

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource_test.go

@@ -16,7 +16,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cespare/xxhash/v2"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -4009,6 +4008,8 @@ func TestGraphQLDataSource(t *testing.T) {
 						NewGraphQLSubscriptionClient(http.DefaultClient, http.DefaultClient, ctx),
 					},
 					PostProcessing: DefaultPostProcessingConfiguration,
+					SourceName:     "ds-id",
+					SourceID:       "ds-id",
 				},
 				Response: &resolve.GraphQLResponse{
 					Fetches: resolve.Sequence(),
@@ -4050,6 +4051,8 @@ func TestGraphQLDataSource(t *testing.T) {
 					client: NewGraphQLSubscriptionClient(http.DefaultClient, http.DefaultClient, ctx),
 				},
 				PostProcessing: DefaultPostProcessingConfiguration,
+				SourceName:     "ds-id",
+				SourceID:       "ds-id",
 			},
 			Response: &resolve.GraphQLResponse{
 				Data: &resolve.Object{
@@ -8246,10 +8249,6 @@ func (f *FailingSubscriptionClient) Subscribe(ctx *resolve.Context, options Grap
 	return errSubscriptionClientFail
 }
 
-func (f *FailingSubscriptionClient) UniqueRequestID(ctx *resolve.Context, options GraphQLSubscriptionOptions, hash *xxhash.Digest) (err error) {
-	return errSubscriptionClientFail
-}
-
 type testSubscriptionUpdaterChan struct {
 	updates  chan string
 	complete chan struct{}
@@ -8441,13 +8440,13 @@ func TestSubscriptionSource_Start(t *testing.T) {
 
 	t.Run("should return error when input is invalid", func(t *testing.T) {
 		source := SubscriptionSource{client: &FailingSubscriptionClient{}}
-		err := source.Start(resolve.NewContext(context.Background()), []byte(`{"url": "", "body": "", "header": null}`), nil)
+		err := source.Start(resolve.NewContext(context.Background()), nil, []byte(`{"url": "", "body": "", "header": null}`), nil)
 		assert.Error(t, err)
 	})
 
 	t.Run("should return error when subscription client returns an error", func(t *testing.T) {
 		source := SubscriptionSource{client: &FailingSubscriptionClient{}}
-		err := source.Start(resolve.NewContext(context.Background()), []byte(`{"url": "", "body": {}, "header": null}`), nil)
+		err := source.Start(resolve.NewContext(context.Background()), nil, []byte(`{"url": "", "body": {}, "header": null}`), nil)
 		assert.Error(t, err)
 		assert.Equal(t, resolve.ErrUnableToResolve, err)
 	})
@@ -8460,7 +8459,7 @@ func TestSubscriptionSource_Start(t *testing.T) {
 
 		source := newSubscriptionSource(ctx.Context())
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: "#test") { text createdBy } }"}`)
-		err := source.Start(ctx, chatSubscriptionOptions, updater)
+		err := source.Start(ctx, nil, chatSubscriptionOptions, updater)
 		require.ErrorIs(t, err, resolve.ErrUnableToResolve)
 	})
 
@@ -8472,7 +8471,7 @@ func TestSubscriptionSource_Start(t *testing.T) {
 
 		source := newSubscriptionSource(ctx.Context())
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomNam: \"#test\") { text createdBy } }"}`)
-		err := source.Start(ctx, chatSubscriptionOptions, updater)
+		err := source.Start(ctx, nil, chatSubscriptionOptions, updater)
 		require.NoError(t, err)
 		updater.AwaitUpdates(t, time.Second, 1)
 		assert.Len(t, updater.updates, 1)
@@ -8490,7 +8489,7 @@ func TestSubscriptionSource_Start(t *testing.T) {
 		source := newSubscriptionSource(resolverLifecycle)
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`)
 
-		err := source.Start(resolve.NewContext(subscriptionLifecycle), chatSubscriptionOptions, updater)
+		err := source.Start(resolve.NewContext(subscriptionLifecycle), nil, chatSubscriptionOptions, updater)
 		require.NoError(t, err)
 
 		username := "myuser"
@@ -8513,7 +8512,7 @@ func TestSubscriptionSource_Start(t *testing.T) {
 		source := newSubscriptionSource(ctx.Context())
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`)
 
-		err := source.Start(ctx, chatSubscriptionOptions, updater)
+		err := source.Start(ctx, nil, chatSubscriptionOptions, updater)
 		require.NoError(t, err)
 
 		username := "myuser"
@@ -8577,7 +8576,7 @@ func TestSubscription_GTWS_SubProtocol(t *testing.T) {
 
 		source := newSubscriptionSource(ctx.Context())
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomNam: \"#test\") { text createdBy } }"}`)
-		err := source.Start(ctx, chatSubscriptionOptions, updater)
+		err := source.Start(ctx, nil, chatSubscriptionOptions, updater)
 		require.NoError(t, err)
 
 		updater.AwaitUpdates(t, time.Second, 1)
@@ -8597,7 +8596,7 @@ func TestSubscription_GTWS_SubProtocol(t *testing.T) {
 		source := newSubscriptionSource(resolverLifecycle)
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`)
 
-		err := source.Start(resolve.NewContext(subscriptionLifecycle), chatSubscriptionOptions, updater)
+		err := source.Start(resolve.NewContext(subscriptionLifecycle), nil, chatSubscriptionOptions, updater)
 		require.NoError(t, err)
 
 		username := "myuser"
@@ -8621,7 +8620,7 @@ func TestSubscription_GTWS_SubProtocol(t *testing.T) {
 		source := newSubscriptionSource(ctx.Context())
 		chatSubscriptionOptions := chatServerSubscriptionOptions(t, `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`)
 
-		err := source.Start(ctx, chatSubscriptionOptions, updater)
+		err := source.Start(ctx, nil, chatSubscriptionOptions, updater)
 		require.NoError(t, err)
 
 		username := "myuser"
@@ -8759,10 +8758,9 @@ func TestSource_Load(t *testing.T) {
 			input = httpclient.SetInputBodyWithPath(input, variables, "variables")
 			input = httpclient.SetInputURL(input, []byte(serverUrl))
 
-			buf := bytes.NewBuffer(nil)
-
-			require.NoError(t, src.Load(context.Background(), input, buf))
-			assert.Equal(t, `{"variables":{"a":null,"b":"b","c":{}}}`, buf.String())
+			data, err := src.Load(context.Background(), nil, input)
+			require.NoError(t, err)
+			assert.Equal(t, `{"variables":{"a":null,"b":"b","c":{}}}`, string(data))
 		})
 	})
 	t.Run("remove undefined variables", func(t *testing.T) {
@@ -8775,16 +8773,16 @@ func TestSource_Load(t *testing.T) {
 			var input []byte
 			input = httpclient.SetInputBodyWithPath(input, variables, "variables")
 			input = httpclient.SetInputURL(input, []byte(serverUrl))
-			buf := bytes.NewBuffer(nil)
 
 			undefinedVariables := []string{"a", "c"}
 			ctx := context.Background()
 			var err error
 			input, err = httpclient.SetUndefinedVariables(input, undefinedVariables)
 			assert.NoError(t, err)
 
-			require.NoError(t, src.Load(ctx, input, buf))
-			assert.Equal(t, `{"variables":{"b":null}}`, buf.String())
+			data, err := src.Load(ctx, nil, input)
+			require.NoError(t, err)
+			assert.Equal(t, `{"variables":{"b":null}}`, string(data))
 		})
 	})
 }
@@ -8866,10 +8864,12 @@ func TestLoadFiles(t *testing.T) {
 		input = httpclient.SetInputBodyWithPath(input, variables, "variables")
 		input = httpclient.SetInputBodyWithPath(input, query, "query")
 		input = httpclient.SetInputURL(input, []byte(serverUrl))
-		buf := bytes.NewBuffer(nil)
 
 		ctx := context.Background()
-		require.NoError(t, src.LoadWithFiles(ctx, input, []*httpclient.FileUpload{httpclient.NewFileUpload(f.Name(), fileName, "variables.file")}, buf))
+		got, err := src.LoadWithFiles(ctx, nil, input, []*httpclient.FileUpload{httpclient.NewFileUpload(f.Name(), fileName, "variables.file")})
+		require.NoError(t, err)
+		require.Equal(t, []byte{}, got)
+
 	})
 
 	t.Run("multiple files", func(t *testing.T) {
@@ -8910,7 +8910,6 @@ func TestLoadFiles(t *testing.T) {
 		input = httpclient.SetInputBodyWithPath(input, variables, "variables")
 		input = httpclient.SetInputBodyWithPath(input, query, "query")
 		input = httpclient.SetInputURL(input, []byte(serverUrl))
-		buf := bytes.NewBuffer(nil)
 
 		dir := t.TempDir()
 		f1, err := os.CreateTemp(dir, file1Name)
@@ -8924,11 +8923,12 @@ func TestLoadFiles(t *testing.T) {
 		assert.NoError(t, err)
 
 		ctx := context.Background()
-		require.NoError(t, src.LoadWithFiles(ctx, input,
+		got, err := src.LoadWithFiles(ctx, nil, input,
 			[]*httpclient.FileUpload{
 				httpclient.NewFileUpload(f1.Name(), file1Name, "variables.files.0"),
-				httpclient.NewFileUpload(f2.Name(), file2Name, "variables.files.1")},
-			buf))
+				httpclient.NewFileUpload(f2.Name(), file2Name, "variables.files.1")})
+		require.NoError(t, err)
+		require.Equal(t, []byte{}, got)
 	})
 }