refactor(search): use errgroup for concurrent page fetch

Replaces manual goroutine and WaitGroup management in fetchAllPages with
golang.org/x/sync/errgroup for improved error handling and concurrency
control. Updates usage and warning messages to clarify concurrency
implications. Adds errgroup dependency to go.mod and go.sum.

Author: Ajnasz

go.mod

@@ -3,3 +3,5 @@ module github.com/Ajnasz/go-loggly-cli
 go 1.25
 
 require github.com/bitly/go-simplejson v0.5.1
+
+require golang.org/x/sync v0.17.0 // indirect

go.sum

@@ -1,2 +1,4 @@
 github.com/bitly/go-simplejson v0.5.1 h1:xgwPbetQScXt1gh9BmoJ6j9JMr3TElvuIyjR8pgdoow=
 github.com/bitly/go-simplejson v0.5.1/go.mod h1:YOPVLzCfwK14b4Sff3oP1AmGhI9T9Vsg84etUnlyp+Q=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=

loggly.go

@@ -13,13 +13,8 @@ import (
 	"github.com/Ajnasz/go-loggly-cli/search"
 )
 
-// Version is the version string
 var version string
 
-//
-// Usage information.
-//
-
 const usage = `
   Usage: loggly [options] [query...]
 
@@ -33,7 +28,7 @@ const usage = `
     -count            print total event count
     -all              print the entire loggly event instead of just the message
     -maxPages <count> maximum number of pages to query [3]
-    -concurrency <count> number of concurrent page fetchers [3]
+    -concurrency <count> number of concurrent page fetchers [3]. If loggly returns with http error consider reducing this value.
     -version          print version information
 
   Operators:
@@ -202,7 +197,7 @@ func warnInvalidFlagPlacement(args []string) {
 
 func warnHighConcurrency(concurrency int) {
 	if concurrency > 3 {
-		fmt.Fprintf(os.Stderr, "Warning: High concurrency (%d) may lead to rate limiting or temporary blocking by Loggly. Consider reducing the concurrency level.\n", concurrency)
+		fmt.Fprintf(os.Stderr, "Warning: High concurrency (%d) may lead to rate limiting or temporary blocking by Loggly. If loggly returns with error, consider reducing the concurrency level.\n", concurrency)
 	}
 }
 

search/search.go

@@ -6,12 +6,12 @@ import (
 	"io"
 	"net/http"
 	"strconv"
-	"sync"
 	"sync/atomic"
 
 	"github.com/Ajnasz/go-loggly-cli/orderedbuffer"
 	"github.com/Ajnasz/go-loggly-cli/semaphore"
 	"github.com/bitly/go-simplejson"
+	"golang.org/x/sync/errgroup"
 )
 
 // Client Loggly search client with user credentials, loggly
@@ -161,14 +161,12 @@ func shouldStopFetching(err error, res *Response, pageSize int) bool {
 	return false
 }
 
-func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Response, errChan chan error) {
+func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Response) error {
 	defer close(resChan)
-	defer close(errChan)
 	j, err := c.CreateSearch(ctx, q.String())
 
 	if err != nil {
-		errChan <- err
-		return
+		return err
 	}
 
 	concurrent := min(q.maxPages, c.concurrency.Load())
@@ -177,37 +175,27 @@ func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Respon
 	var page atomic.Int64
 	page.Store(-1)
 
-	var wg sync.WaitGroup
 	var hasMore atomic.Bool
 	hasMore.Store(true)
 	responsesStore := orderedbuffer.NewOrderedBuffer(resChan)
 
+	errg, ctx := errgroup.WithContext(ctx)
+
 	for {
 		if err := sem.Acquire(ctx); err != nil {
-			errChan <- err
-			break
+			return err
 		}
-
-		wg.Add(1)
-		go func(page int) {
-			defer wg.Done()
+		p := int(page.Add(1))
+		errg.Go(func() error {
 			defer sem.Release()
 
-			if ctx.Err() != nil {
-				errChan <- ctx.Err()
-				hasMore.Store(false)
-				return
-			}
-
-			res, err := c.fetchAndStorePage(ctx, j, responsesStore, page)
+			res, err := c.fetchAndStorePage(ctx, j, responsesStore, p)
 
 			if shouldStopFetching(err, res, q.size) {
 				hasMore.Store(false)
-				if err != nil {
-					errChan <- err
-				}
 			}
-		}(int(page.Add(1)))
+			return err
+		})
 
 		shouldBreak := page.Load() >= q.maxPages || !hasMore.Load()
 
@@ -216,7 +204,7 @@ func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Respon
 		}
 	}
 
-	wg.Wait()
+	return errg.Wait()
 }
 
 // Fetch Search response with total events, page number
@@ -229,7 +217,13 @@ func (c *Client) Fetch(ctx context.Context, q Query) (chan Response, chan error)
 	resChan := make(chan Response)
 	errChan := make(chan error)
 
-	go c.fetchAllPages(ctx, q, resChan, errChan)
+	go func() {
+		defer close(errChan)
+		err := c.fetchAllPages(ctx, q, resChan)
+		if err != nil {
+			errChan <- err
+		}
+	}()
 
 	return resChan, errChan
 }