feat: singleflight to deduplicate GET requests

Author: ndk

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,24 +106,28 @@ 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.
 type OperationIDKey struct{}
 
-func (d *aivenClient) Do(ctx context.Context, operationID, method, path string, in any, query ...[2]string) ([]byte, error) {
+func (d *aivenClient) Do(ctx context.Context, operationID, method, path string, in any, query ...[2]string) (_ []byte, err error) {
 	ctx = context.WithValue(ctx, OperationIDKey{}, operationID)
-	var rsp *http.Response
-	var err error
+	queryString := fmtQuery(operationID, query...)
+
+	var statusCode int
+	var shared bool
 
 	if d.Debug {
 		start := time.Now()
@@ -133,54 +138,75 @@ func (d *aivenClient) Do(ctx context.Context, operationID, method, path string,
 			if err != nil {
 				event = d.logger.Error().Err(err)
 			} else {
-				event = d.logger.Info().Str("status", rsp.Status)
+				event = d.logger.Info()
 			}
 
 			event.Ctx(ctx).
 				Stringer("duration", end).
 				Str("operationID", operationID).
 				Str("method", method).
 				Str("path", path).
-				Str("query", fmtQuery(operationID, query...)).
+				Str("query", queryString).
+				Int("status_code", statusCode).
+				Bool("shared", shared).
 				Send()
 		}()
 	}
 
-	rsp, err = d.do(ctx, operationID, method, path, in, query...)
+	var body []byte
+	if d.EnableSingleFlight && (method == http.MethodGet || method == http.MethodHead || method == http.MethodOptions || method == http.MethodTrace) {
+		type result struct {
+			statusCode int
+			body       []byte
+		}
+		key := strings.Join([]string{method, d.Host, path, queryString}, "|")
+		v, serr, sh := d.singleflight.Do(key, func() (any, error) {
+			statusCode, body, err := d.do(ctx, method, path, in, queryString)
+			return result{statusCode: statusCode, body: body}, err
+		})
+		res := v.(result)
+		statusCode, body, err, shared = res.statusCode, res.body, serr, sh
+	} else {
+		statusCode, body, err = d.do(ctx, method, path, in, queryString)
+	}
 	if err != nil {
 		return nil, err
 	}
 
-	defer func() {
-		err = multierror.Append(rsp.Body.Close()).ErrorOrNil()
-	}()
-
-	return fromResponse(operationID, rsp)
+	return fromBytes(operationID, statusCode, body)
 }
 
-func (d *aivenClient) do(ctx context.Context, operationID, method, path string, in any, query ...[2]string) (*http.Response, error) {
+func (d *aivenClient) do(ctx context.Context, method, path string, in any, queryString string) (int, []byte, error) {
 	var body io.Reader
 
 	if !(in == nil || isEmpty(in)) {
 		b, err := json.Marshal(in)
 		if err != nil {
-			return nil, err
+			return 0, nil, err
 		}
 
 		body = bytes.NewBuffer(b)
 	}
 
 	req, err := http.NewRequestWithContext(ctx, method, d.Host+path, body)
 	if err != nil {
-		return nil, err
+		return 0, nil, err
 	}
 
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("User-Agent", d.UserAgent)
 	req.Header.Set("Authorization", "aivenv1 "+d.Token)
-	req.URL.RawQuery = fmtQuery(operationID, query...)
+	req.URL.RawQuery = queryString
 
-	return d.doer.Do(req)
+	rsp, err := d.doer.Do(req)
+	if err != nil {
+		return 0, nil, err
+	}
+	defer func() {
+		err = multierror.Append(rsp.Body.Close()).ErrorOrNil()
+	}()
+	respBody, err := io.ReadAll(rsp.Body)
+	return rsp.StatusCode, respBody, err
 }
 
 func isEmpty(a any) bool {

client_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -299,3 +300,151 @@ 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 := t.Context()
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 100)
+	for range 100 {
+		wg.Go(func() {
+			_, err := c.ServiceList(ctx, "test-project")
+			errs <- err
+		})
+	}
+	wg.Wait()
+	close(errs)
+
+	for err := range errs {
+		require.NoError(t, err)
+	}
+
+	// We expect less than 10 calls to the API because of the single flight deduplication.
+	assert.Less(t, callCount, int64(10))
+}
+
+// TestSingleFlightDifferentProjects tests that singleflight doesn't merge different paths.
+func TestSingleFlightDifferentProjects(t *testing.T) {
+	var callCount int64
+
+	mux := http.NewServeMux()
+	mux.HandleFunc(
+		"/v1/project/project-a/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)
+		},
+	)
+	mux.HandleFunc(
+		"/v1/project/project-b/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 := t.Context()
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 2)
+	wg.Go(func() {
+		_, err := c.ServiceList(ctx, "project-a")
+		errs <- err
+	})
+	wg.Go(func() {
+		_, err := c.ServiceList(ctx, "project-b")
+		errs <- err
+	})
+	wg.Wait()
+	close(errs)
+
+	for err := range errs {
+		require.NoError(t, err)
+	}
+
+	// We expect exactly 2 calls to the API because different project names should not be deduplicated.
+	assert.Equal(t, int64(2), callCount)
+}
+
+func TestSingleFlightMethodIsolation(t *testing.T) {
+	var callCount int64
+
+	mux := http.NewServeMux()
+	mux.HandleFunc(
+		"/v1/test",
+		func(w http.ResponseWriter, _ *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			_, err := w.Write([]byte(`{}`))
+			assert.NoError(t, err)
+			atomic.AddInt64(&callCount, 1)
+		},
+	)
+
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	d := &aivenClient{
+		Host:               server.URL,
+		UserAgent:          "unit-test",
+		Token:              "token",
+		EnableSingleFlight: true,
+		doer:               server.Client(),
+	}
+
+	ctx := t.Context()
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 2)
+	wg.Go(func() {
+		_, err := d.Do(ctx, "op-get", http.MethodGet, "/v1/test", nil)
+		errs <- err
+	})
+	wg.Go(func() {
+		_, err := d.Do(ctx, "op-head", http.MethodHead, "/v1/test", nil)
+		errs <- err
+	})
+	wg.Wait()
+	close(errs)
+
+	for err := range errs {
+		require.NoError(t, err)
+	}
+
+	// We expect exactly 2 calls to the API because different methods should not be deduplicated.
+	assert.Equal(t, int64(2), callCount)
+}

error.go

@@ -55,19 +55,14 @@ func IsAlreadyExists(err error) bool {
 	return errors.As(err, &e) && strings.Contains(e.Message, "already exists") && e.Status == http.StatusConflict
 }
 
-func fromResponse(operationID string, rsp *http.Response) ([]byte, error) {
-	e := Error{OperationID: operationID, Status: rsp.StatusCode}
-	b, err := io.ReadAll(rsp.Body)
-	if err != nil {
-		e.Message = fmt.Sprintf("body read error: %s", err)
-		return nil, e
-	}
+func fromBytes(operationID string, statusCode int, b []byte) ([]byte, error) {
+	e := Error{OperationID: operationID, Status: statusCode}
 
-	if rsp.StatusCode < 200 || rsp.StatusCode >= 300 {
+	if statusCode < 200 || statusCode >= 300 {
 		// According to the documentation,
 		// failed responses must have "errors" and "message" fields
 		// https://api.aiven.io/doc/#section/Responses/Failed-responses
-		err = json.Unmarshal(b, &e)
+		err := json.Unmarshal(b, &e)
 		if err != nil {
 			// 1. The body might contain sensitive data
 			// 2. It might fail the unmarshalling into Error and still be a valid json
@@ -84,3 +79,14 @@ func fromResponse(operationID string, rsp *http.Response) ([]byte, error) {
 
 	return b, nil
 }
+
+func fromResponse(operationID string, rsp *http.Response) ([]byte, error) {
+	b, err := io.ReadAll(rsp.Body)
+	if err != nil {
+		e := Error{OperationID: operationID, Status: rsp.StatusCode}
+		e.Message = fmt.Sprintf("body read error: %s", err)
+		return nil, e
+	}
+
+	return fromBytes(operationID, rsp.StatusCode, b)
+}

go.mod

@@ -24,6 +24,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/samber/lo v1.52.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
 	golang.org/x/text v0.22.0 // indirect
 )

go.sum

@@ -43,6 +43,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
+	}
+}