chore: add comments

Author: jensneuse

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

@@ -2437,7 +2437,7 @@ func TestWebSocketUpgradeFailures(t *testing.T) {
 					w.Header().Set(key, value)
 				}
 				w.WriteHeader(tc.statusCode)
-				fmt.Fprintf(w, `{"error": "WebSocket upgrade failed", "status": %d}`, tc.statusCode)
+				_, _ = fmt.Fprintf(w, `{"error": "WebSocket upgrade failed", "status": %d}`, tc.statusCode)
 			}))
 			defer server.Close()
 

v2/pkg/engine/datasource/httpclient/nethttpclient.go

@@ -136,6 +136,9 @@ const (
 	sizeHintKey httpClientContext = "size-hint"
 )
 
+// WithHTTPClientSizeHint allows the engine to keep track of response sizes per subgraph fetch
+// If a hint is supplied, we can create a buffer of size close to the required size
+// This reduces allocations by reducing the buffer grow calls, which always copies the buffer
 func WithHTTPClientSizeHint(ctx context.Context, size int) context.Context {
 	return context.WithValue(ctx, sizeHintKey, size)
 }
@@ -144,6 +147,9 @@ func buffer(ctx context.Context) *bytes.Buffer {
 	if sizeHint, ok := ctx.Value(sizeHintKey).(int); ok && sizeHint > 0 {
 		return bytes.NewBuffer(make([]byte, 0, sizeHint))
 	}
+	// if we start with zero, doubling will take a while until we reach the required size
+	// if we start with a high number, e.g. 1024, we just increase the memory usage of the engine
+	// 64 seems to be a healthy middle ground
 	return bytes.NewBuffer(make([]byte, 0, 64))
 }
 
@@ -211,6 +217,8 @@ func makeHTTPRequest(client *http.Client, ctx context.Context, baseHeaders http.
 	request.Header.Set(AcceptEncodingHeader, EncodingGzip)
 	request.Header.Add(AcceptEncodingHeader, EncodingDeflate)
 	if contentLength > 0 {
+		// always set the Content-Length Header so that chunking can be avoided
+		// and other parties can more efficiently parse
 		request.Header.Set(ContentLengthHeader, fmt.Sprintf("%d", contentLength))
 	}
 
@@ -229,6 +237,12 @@ func makeHTTPRequest(client *http.Client, ctx context.Context, baseHeaders http.
 		return nil, err
 	}
 
+	// we intentionally don't use a pool of sorts here
+	// we're buffering the response and then later, in the engine,
+	// parse it into an JSON AST with the use of an arena, which is quite efficient
+	// Through trial and error it turned out that it's best to leave this buffer to the GC
+	// It'll know best the lifecycle of the buffer
+	// Using an arena here just increased overall memory usage
 	out := buffer(ctx)
 	_, err = out.ReadFrom(respReader)
 	if err != nil {

v2/pkg/engine/resolve/context.go

@@ -36,10 +36,17 @@ type Context struct {
 	SubgraphHeadersBuilder SubgraphHeadersBuilder
 }
 
+// SubgraphHeadersBuilder allows the user of the engine to "define" the headers for a subgraph request
+// Instead of going back and forth between engine & transport,
+// you can simply define a function that returns headers for a Subgraph request
+// In addition to just the header, the implementer can return a hash for the header which will be used by request deduplication
 type SubgraphHeadersBuilder interface {
+	// HeadersForSubgraph must return the headers and a hash for a Subgraph Request
+	// The hash will be used for request deduplication
 	HeadersForSubgraph(subgraphName string) (http.Header, uint64)
 }
 
+// HeadersForSubgraphRequest returns headers and a hash for a request that the engine will make to a subgraph
 func (c *Context) HeadersForSubgraphRequest(subgraphName string) (http.Header, uint64) {
 	if c.SubgraphHeadersBuilder == nil {
 		return nil, 0

v2/pkg/engine/resolve/loader.go

@@ -137,8 +137,9 @@ type result struct {
 	loaderHookContext context.Context
 
 	httpResponseContext *httpclient.ResponseContext
-	out                 []byte
-	singleFlightStats   *singleFlightStats
+	// out is the subgraph response body
+	out               []byte
+	singleFlightStats *singleFlightStats
 }
 
 func (r *result) init(postProcessing PostProcessingConfiguration, info *FetchInfo) {
@@ -182,6 +183,14 @@ type Loader struct {
 
 	taintedObjs taintedObjects
 
+	// jsonArena is the arena to allocation json, supplied by the Resolver
+	// Disclaimer: this arena is NOT thread safe!
+	// Only use from main goroutine
+	// Don't Reset or Release, the Resolver handles this
+	// Disclaimer: When parsing json into the arena, the underlying bytes must also be allocated on the arena!
+	// This is very important to "tie" their lifecycles together
+	// If you're not doing this, you will see segfaults
+	// Example of correct usage in func "mergeResult"
 	jsonArena arena.Arena
 	sf        *SingleFlight
 }
@@ -773,9 +782,11 @@ func (l *Loader) mergeErrors(res *result, fetchItem *FetchItem, value *astjson.V
 				return err
 			}
 		}
-
-		// If the error propagation mode is pass-through, we append the errors to the root array
+		// for efficiency purposes, resolvable.errors is not initialized
+		// don't change this, it's measurable
+		// downside: we have to verify it's initialized before appending to it
 		l.resolvable.ensureErrorsInitialized()
+		// If the error propagation mode is pass-through, we append the errors to the root array
 		l.resolvable.errors.AppendArrayItems(value)
 		return nil
 	}
@@ -811,7 +822,9 @@ func (l *Loader) mergeErrors(res *result, fetchItem *FetchItem, value *astjson.V
 	if err := l.addApolloRouterCompatibilityError(res); err != nil {
 		return err
 	}
-
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	astjson.AppendToArray(l.resolvable.errors, errorObject)
 
@@ -1066,7 +1079,9 @@ func (l *Loader) addApolloRouterCompatibilityError(res *result) error {
 	if err != nil {
 		return err
 	}
-
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	astjson.AppendToArray(l.resolvable.errors, apolloRouterStatusError)
 
@@ -1081,6 +1096,9 @@ func (l *Loader) renderErrorsFailedDeps(fetchItem *FetchItem, res *result) error
 		return err
 	}
 	l.setSubgraphStatusCode([]*astjson.Value{errorObject}, res.statusCode)
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	astjson.AppendToArray(l.resolvable.errors, errorObject)
 	return nil
@@ -1093,6 +1111,9 @@ func (l *Loader) renderErrorsFailedToFetch(fetchItem *FetchItem, res *result, re
 		return err
 	}
 	l.setSubgraphStatusCode([]*astjson.Value{errorObject}, res.statusCode)
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	astjson.AppendToArray(l.resolvable.errors, errorObject)
 	return nil
@@ -1112,6 +1133,9 @@ func (l *Loader) renderErrorsStatusFallback(fetchItem *FetchItem, res *result, s
 	}
 
 	l.setSubgraphStatusCode([]*astjson.Value{errorObject}, res.statusCode)
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	astjson.AppendToArray(l.resolvable.errors, errorObject)
 	return nil
@@ -1137,6 +1161,9 @@ func (l *Loader) renderAuthorizationRejectedErrors(fetchItem *FetchItem, res *re
 	}
 	pathPart := l.renderAtPathErrorPart(fetchItem.ResponsePath)
 	extensionErrorCode := fmt.Sprintf(`"extensions":{"code":"%s"}`, errorcodes.UnauthorizedFieldOrType)
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	if res.ds.Name == "" {
 		for _, reason := range res.authorizationRejectedReasons {
@@ -1216,6 +1243,9 @@ func (l *Loader) renderRateLimitRejectedErrors(fetchItem *FetchItem, res *result
 			return err
 		}
 	}
+	// for efficiency purposes, resolvable.errors is not initialized
+	// don't change this, it's measurable
+	// downside: we have to verify it's initialized before appending to it
 	l.resolvable.ensureErrorsInitialized()
 	astjson.AppendToArray(l.resolvable.errors, errorObject)
 	return nil
@@ -1417,7 +1447,7 @@ func (l *Loader) loadBatchEntityFetch(ctx context.Context, fetchItem *FetchItem,
 			}
 		}
 	}
-
+	// I tried using arena here but it only worsened the situation
 	preparedInput := bytes.NewBuffer(make([]byte, 0, 64))
 	itemInput := bytes.NewBuffer(make([]byte, 0, 32))
 	keyGen := pool.Hash64.Get()
@@ -1579,6 +1609,7 @@ const (
 	operationTypeContextKey loaderContextKey = "operationType"
 )
 
+// GetOperationTypeFromContext can be used, e.g. by the transport, to check if the operation is a Mutation
 func GetOperationTypeFromContext(ctx context.Context) ast.OperationType {
 	if ctx == nil {
 		return ast.OperationTypeQuery
@@ -1638,6 +1669,7 @@ func (l *Loader) loadByContext(ctx context.Context, source DataSource, fetchItem
 		return nil
 	}
 
+	// helps the http client to create buffers at the right size
 	ctx = httpclient.WithHTTPClientSizeHint(ctx, item.sizeHint)
 
 	defer l.sf.Finish(sfKey, fetchKey, item)
@@ -1851,6 +1883,9 @@ func (l *Loader) compactJSON(data []byte) ([]byte, error) {
 		return nil, err
 	}
 	out := dst.Bytes()
+	// don't use arena here or segfault
+	// it's also not a hot path and not important to optimize
+	// arena requires the parsed content to be on the arena as well
 	v, err := astjson.ParseBytes(out)
 	if err != nil {
 		return nil, err

v2/pkg/engine/resolve/resolvable.go

@@ -31,7 +31,8 @@ type Resolvable struct {
 	errors               *astjson.Value
 	valueCompletion      *astjson.Value
 	skipAddingNullErrors bool
-
+	// astjsonArena is the arena to handle json, supplied by Resolver
+	// not thread safe, but Resolvable is single threaded anyways
 	astjsonArena arena.Arena
 	parsers      []*astjson.Parser
 
@@ -111,6 +112,7 @@ func (r *Resolvable) Init(ctx *Context, initialData []byte, operationType ast.Op
 	r.operationType = operationType
 	r.renameTypeNames = ctx.RenameTypeNames
 	r.data = astjson.ObjectValue(r.astjsonArena)
+	// don't init errors! It will heavily increase memory usage
 	r.errors = nil
 	if initialData != nil {
 		initialValue, err := astjson.ParseBytesWithArena(r.astjsonArena, initialData)
@@ -129,6 +131,7 @@ func (r *Resolvable) InitSubscription(ctx *Context, initialData []byte, postProc
 	r.ctx = ctx
 	r.operationType = ast.OperationTypeSubscription
 	r.renameTypeNames = ctx.RenameTypeNames
+	// don't init errors! It will heavily increase memory usage
 	r.errors = nil
 	if initialData != nil {
 		initialValue, err := astjson.ParseBytesWithArena(r.astjsonArena, initialData)
@@ -167,6 +170,7 @@ func (r *Resolvable) ResolveNode(node Node, data *astjson.Value, out io.Writer)
 	r.print = false
 	r.printErr = nil
 	r.authorizationError = nil
+	// don't init errors! It will heavily increase memory usage
 	r.errors = nil
 
 	hasErrors := r.walkNode(node, data)
@@ -233,6 +237,7 @@ func (r *Resolvable) Resolve(ctx context.Context, rootData *Object, fetchTree *F
 	return r.printErr
 }
 
+// ensureErrorsInitialized is used to lazily init r.errors if needed
 func (r *Resolvable) ensureErrorsInitialized() {
 	if r.errors == nil {
 		r.errors = astjson.ArrayValue(r.astjsonArena)

v2/pkg/engine/resolve/resolve.go

@@ -72,7 +72,13 @@ type Resolver struct {
 	// maxSubscriptionFetchTimeout defines the maximum time a subscription fetch can take before it is considered timed out
 	maxSubscriptionFetchTimeout time.Duration
 
-	resolveArenaPool   *ArenaPool
+	// resolveArenaPool is the arena pool dedicated for Loader & Resolvable
+	// ArenaPool automatically adjusts arena buffer sizes per workload
+	// resolving & response buffering are very different tasks
+	// as such, it was best to have two arena pools in terms of memory usage
+	// A single pool for both was much less efficient
+	resolveArenaPool *ArenaPool
+	// responseBufferPool is the arena pool dedicated for response buffering before sending to the client
 	responseBufferPool *ArenaPool
 
 	// Single flight cache for deduplicating requests across all loaders
@@ -246,6 +252,7 @@ func New(ctx context.Context, options ResolverOptions) *Resolver {
 
 func newTools(options ResolverOptions, allowedExtensionFields map[string]struct{}, allowedErrorFields map[string]struct{}, sf *SingleFlight) *tools {
 	return &tools{
+		// we set the arena manually
 		resolvable: NewResolvable(nil, options.ResolvableOptions),
 		loader: &Loader{
 			propagateSubgraphErrors:                      options.PropagateSubgraphErrors,