feat: singleflight to deduplicate GET requests

Resolves NEX-2296.

Add singleflight to deduplicate concurrent GET/HEAD/OPTIONS/TRACE requests.

Deduplicate concurrent requests to shared list endpoints using singleflight.
Many resources, such as databases,
do not have individual endpoints and instead share a single "list" endpoint (ServiceDatabaseList).
This often results in multiple simultaneous requests for the same data,
which can cause redundant API calls and unnecessary load.
Currently, the client uses only GET for reading,
but additional methods may be supported in the future.

Author: byashimov

README.md

@@ -21,6 +21,8 @@ go get github.com/aiven/go-client-codegen
 | `AIVEN_USER_AGENT` | `string` | User Agent                     |
 | `AIVEN_DEBUG`      | `bool`   | Debug Output Flag (stderr)     |
 
+See all configuration options in [`client.go`](client.go).
+
 #### Via Constructor Options
 
 ```go

client.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"golang.org/x/exp/slices"
+	"golang.org/x/sync/singleflight"
 
 	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/go-retryablehttp"
@@ -105,15 +106,17 @@ func NewClient(opts ...Option) (Client, error) {
 // Default values: 26 seconds
 // Changed values: 67 seconds
 type aivenClient struct {
-	Host         string        `envconfig:"AIVEN_WEB_URL" default:"https://api.aiven.io"`
-	UserAgent    string        `envconfig:"AIVEN_USER_AGENT" default:"aiven-go-client/v3"`
-	Token        string        `envconfig:"AIVEN_TOKEN"`
-	Debug        bool          `envconfig:"AIVEN_DEBUG"`
-	RetryMax     int           `envconfig:"AIVEN_CLIENT_RETRY_MAX" default:"6"`
-	RetryWaitMin time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MIN" default:"2s"`
-	RetryWaitMax time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MAX" default:"15s"`
-	logger       zerolog.Logger
-	doer         Doer
+	Host               string        `envconfig:"AIVEN_WEB_URL" default:"https://api.aiven.io"`
+	UserAgent          string        `envconfig:"AIVEN_USER_AGENT" default:"aiven-go-client/v3"`
+	Token              string        `envconfig:"AIVEN_TOKEN"`
+	Debug              bool          `envconfig:"AIVEN_DEBUG"`
+	RetryMax           int           `envconfig:"AIVEN_CLIENT_RETRY_MAX" default:"6"`
+	RetryWaitMin       time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MIN" default:"2s"`
+	RetryWaitMax       time.Duration `envconfig:"AIVEN_CLIENT_RETRY_WAIT_MAX" default:"15s"`
+	EnableSingleFlight bool          `envconfig:"AIVEN_CLIENT_ENABLE_SINGLE_FLIGHT" default:"true"`
+	logger             zerolog.Logger
+	doer               Doer
+	singleflight       singleflight.Group
 }
 
 // OperationIDKey is the key used to store the operation ID in the context.
@@ -180,6 +183,26 @@ func (d *aivenClient) do(ctx context.Context, operationID, method, path string,
 	req.Header.Set("Authorization", "aivenv1 "+d.Token)
 	req.URL.RawQuery = fmtQuery(operationID, query...)
 
+	// Deduplicate concurrent requests to shared list endpoints using singleflight.
+	// Many resources, such as databases,
+	// do not have individual endpoints and instead share a single "list" endpoint (e.g., ServiceDatabaseList).
+	// This often results in multiple simultaneous requests for the same data,
+	// which can cause redundant API calls and unnecessary load.
+	// Currently, the client uses only GET for reading,
+	// but additional methods may be supported in the future.
+	if d.EnableSingleFlight {
+		switch method {
+		case http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodTrace:
+			v, err, _ := d.singleflight.Do(req.URL.String(), func() (any, error) {
+				return d.doer.Do(req)
+			})
+			if err != nil {
+				return nil, err
+			}
+			return v.(*http.Response), err
+		}
+	}
+
 	return d.doer.Do(req)
 }
 

client_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -299,3 +300,39 @@ func TestServiceIntegrationEndpointGet(t *testing.T) {
 	// All calls are received
 	assert.EqualValues(t, 1, callCount)
 }
+
+func TestSingleFlightDeduplication(t *testing.T) {
+	var callCount int64
+
+	mux := http.NewServeMux()
+	mux.HandleFunc(
+		"/v1/project/test-project/service",
+		func(w http.ResponseWriter, _ *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			_, err := w.Write([]byte(`{"services": []}`))
+			assert.NoError(t, err)
+			atomic.AddInt64(&callCount, 1)
+		},
+	)
+
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	c, err := NewClient(TokenOpt("token"), HostOpt(server.URL), UserAgentOpt("unit-test"))
+	require.NotNil(t, c)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	var wg sync.WaitGroup
+	for range 100 {
+		wg.Go(func() {
+			_, _ = c.ServiceList(ctx, "test-project")
+		})
+	}
+	wg.Wait()
+
+	// We expect less than 10 calls to the API because of the single flight deduplication.
+	assert.Less(t, callCount, int64(10))
+}

go.mod

@@ -23,5 +23,6 @@ require (
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
+	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
 )

go.sum

@@ -41,6 +41,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 h1:hNQpMuAJe5CtcUqCXaWga3FHu+kQvCqcsoVaQgSV60o=
 golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08=
+golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
+golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

option.go

@@ -62,3 +62,10 @@ func RetryWaitMaxOpt(retryWaitMax time.Duration) Option {
 		d.RetryWaitMax = retryWaitMax
 	}
 }
+
+// EnableSingleFlightOpt enables singleflight for deduplicating concurrent identical requests
+func EnableSingleFlightOpt(enableSingleFlight bool) Option {
+	return func(d *aivenClient) {
+		d.EnableSingleFlight = enableSingleFlight
+	}
+}