perf(concurrency): cancel remaining goroutines when has error (#9484)

**Description**

This PR optimizes goroutine execution in multi-task scenarios:

- Uses errgroup.WithContext to manage multiple goroutines and propagate
cancellation signals.
- When any goroutine encounters an error, errgroup automatically cancels
the remaining goroutines, allowing them to exit early.
- Reduces unnecessary computation and resource usage when partial
failures occur.

This improves system responsiveness and reduces wasted CPU cycles in
error cases.

**Checklist**

- [x] Code compiles correctly and linting passes locally
- [ ] For all _code_ changes, an entry added to the `CHANGELOG.md` file
describing and linking to
      this PR
- [ ] Tests added for new functionality, or regression tests for bug
fixes added as applicable

---------

Co-authored-by: Harshil Goel <[email protected]>
Co-authored-by: Matthew McNeely <[email protected]>

Author: 3 people

worker/task.go

@@ -823,9 +823,9 @@ func (qs *queryState) handleUidPostings(
 	needFiltering := needsStringFiltering(srcFn, q.Langs, q.Attr)
 	isList := schema.State().IsList(q.Attr)
 
-	errCh := make(chan error, numGo)
 	outputs := make([]*pb.Result, numGo)
 
+	eg, egCtx := errgroup.WithContext(ctx)
 	calculate := func(start, end int) error {
 		x.AssertTrue(start%width == 0)
 		out := &pb.Result{}
@@ -834,8 +834,8 @@ func (qs *queryState) handleUidPostings(
 		for i := start; i < end; i++ {
 			if i%100 == 0 {
 				select {
-				case <-ctx.Done():
-					return ctx.Err()
+				case <-egCtx.Done():
+					return egCtx.Err()
 				default:
 				}
 			}
@@ -978,14 +978,12 @@ func (qs *queryState) handleUidPostings(
 		if end > srcFn.n {
 			end = srcFn.n
 		}
-		go func(start, end int) {
-			errCh <- calculate(start, end)
-		}(start, end)
+		eg.Go(func() error {
+			return calculate(start, end)
+		})
 	}
-	for range numGo {
-		if err := <-errCh; err != nil {
-			return err
-		}
+	if err := eg.Wait(); err != nil {
+		return err
 	}
 	// All goroutines are done. Now attach their results.
 	out := args.out
@@ -1635,11 +1633,20 @@ func (qs *queryState) filterGeoFunction(ctx context.Context, arg funcArgs) error
 		attribute.Int("num_go", numGo),
 		attribute.Int("width", width)))
 
+	eg, egCtx := errgroup.WithContext(ctx)
 	filtered := make([]*pb.List, numGo)
 	filter := func(idx, start, end int) error {
 		filtered[idx] = &pb.List{}
 		out := filtered[idx]
-		for _, uid := range uids.Uids[start:end] {
+		for i := start; i < end; i++ {
+			uid := uids.Uids[i]
+			if i%100 == 0 {
+				select {
+				case <-egCtx.Done():
+					return egCtx.Err()
+				default:
+				}
+			}
 			pl, err := qs.cache.Get(x.DataKey(attr, uid))
 			if err != nil {
 				return err
@@ -1661,21 +1668,19 @@ func (qs *queryState) filterGeoFunction(ctx context.Context, arg funcArgs) error
 		return nil
 	}
 
-	errCh := make(chan error, numGo)
 	for i := range numGo {
 		start := i * width
 		end := start + width
 		if end > len(uids.Uids) {
 			end = len(uids.Uids)
 		}
-		go func(idx, start, end int) {
-			errCh <- filter(idx, start, end)
-		}(i, start, end)
+		idx := i
+		eg.Go(func() error {
+			return filter(idx, start, end)
+		})
 	}
-	for range numGo {
-		if err := <-errCh; err != nil {
-			return err
-		}
+	if err := eg.Wait(); err != nil {
+		return err
 	}
 	final := &pb.List{}
 	for _, out := range filtered {