Fix race condition in endpoint discovery

CHANGELOG_PENDING.md

@@ -3,3 +3,4 @@
 ### SDK Enhancements
 
 ### SDK Bugs
+* `aws/crr`: Fixed a race condition that caused concurrent calls relying on endpoint discovery to share the same `url.URL` reference in their operation's `http.Request`.

aws/crr/cache.go

@@ -34,7 +34,10 @@ func (c *EndpointCache) get(endpointKey string) (Endpoint, bool) {
  return Endpoint{}, false
  }
 
- c.endpoints.Store(endpointKey, endpoint)
+ ev := endpoint.(Endpoint)
+ ev.Prune()
+
+ c.endpoints.Store(endpointKey, ev)
  return endpoint.(Endpoint), true
 }
 

aws/crr/cache_test.go

@@ -4,6 +4,7 @@ import (
  "net/url"
  "reflect"
  "testing"
+ "time"
 )
 
 func urlParse(uri string) *url.URL {
@@ -450,3 +451,42 @@ func TestCacheGet(t *testing.T) {
  }
  }
 }
+
+func TestEndpointCache_Get_prune(t *testing.T) {
+ c := NewEndpointCache(2)
+ c.Add(Endpoint{
+ Key: "foo",
+ Addresses: []WeightedAddress{
+ {
+ URL: &url.URL{
+ Host: "foo.amazonaws.com",
+ },
+ Expired: time.Now().Add(5 * time.Minute),
+ },
+ {
+ URL: &url.URL{
+ Host: "bar.amazonaws.com",
+ },
+ Expired: time.Now().Add(5 * -time.Minute),
+ },
+ },
+ })
+
+ load, _ := c.endpoints.Load("foo")
+ if ev := load.(Endpoint); len(ev.Addresses) != 2 {
+ t.Errorf("expected two weighted addresses")
+ }
+
+ weightedAddress, err := c.Get(nil, "foo", false)
+ if err != nil {
+ t.Errorf("expect no error, got %v", err)
+ }
+ if e, a := "foo.amazonaws.com", weightedAddress.URL.Host; e != a {
+ t.Errorf("expect %v, got %v", e, a)
+ }
+
+ load, _ = c.endpoints.Load("foo")
+ if ev := load.(Endpoint); len(ev.Addresses) != 1 {
+ t.Errorf("expected one weighted address")
+ }
+}

aws/crr/endpoint.go

@@ -60,12 +60,32 @@ func (e *Endpoint) GetValidAddress() (WeightedAddress, bool) {
  continue
  }
 
+ we.URL = cloneURL(we.URL)
+
  return we, true
  }
 
  return WeightedAddress{}, false
 }
 
+// Prune will prune the expired addresses from the endpoint by allocating a new []WeightAddress.
+// This is not concurrent safe, and should be called from a single owning thread.
+func (e *Endpoint) Prune() bool {
+ validLen := e.Len()
+ if validLen == len(e.Addresses) {
+ return false
+ }
+ wa := make([]WeightedAddress, 0, validLen)
+ for i := range e.Addresses {
+ if e.Addresses[i].HasExpired() {
+ continue
+ }
+ wa = append(wa, e.Addresses[i])
+ }
+ e.Addresses = wa
+ return true
+}
+
 // Discoverer is an interface used to discovery which endpoint hit. This
 // allows for specifics about what parameters need to be used to be contained
 // in the Discoverer implementor.
@@ -97,3 +117,16 @@ func BuildEndpointKey(params map[string]*string) string {
 
  return strings.Join(values, ".")
 }
+
+func cloneURL(u *url.URL) (clone *url.URL) {
+ clone = &url.URL{}
+
+ *clone = *u
+
+ if u.User != nil {
+ user := *u.User
+ clone.User = &user
+ }
+
+ return clone
+}

aws/crr/endpoint_test.go

@@ -0,0 +1,126 @@
+//go:build go1.16
+// +build go1.16
+
+package crr
+
+import (
+ "net/url"
+ "reflect"
+ "strconv"
+ "testing"
+ "time"
+)
+
+func Test_cloneURL(t *testing.T) {
+ tests := []struct {
+ value *url.URL
+ wantClone *url.URL
+ }{
+ {
+ value: &url.URL{
+ Scheme: "https",
+ Opaque: "foo",
+ User: nil,
+ Host: "amazonaws.com",
+ Path: "/",
+ RawPath: "/",
+ ForceQuery: true,
+ RawQuery: "thing=value",
+ Fragment: "1234",
+ RawFragment: "1234",
+ },
+ wantClone: &url.URL{
+ Scheme: "https",
+ Opaque: "foo",
+ User: nil,
+ Host: "amazonaws.com",
+ Path: "/",
+ RawPath: "/",
+ ForceQuery: true,
+ RawQuery: "thing=value",
+ Fragment: "1234",
+ RawFragment: "1234",
+ },
+ },
+ {
+ value: &url.URL{
+ Scheme: "https",
+ Opaque: "foo",
+ User: url.UserPassword("NOT", "VALID"),
+ Host: "amazonaws.com",
+ Path: "/",
+ RawPath: "/",
+ ForceQuery: true,
+ RawQuery: "thing=value",
+ Fragment: "1234",
+ RawFragment: "1234",
+ },
+ wantClone: &url.URL{
+ Scheme: "https",
+ Opaque: "foo",
+ User: url.UserPassword("NOT", "VALID"),
+ Host: "amazonaws.com",
+ Path: "/",
+ RawPath: "/",
+ ForceQuery: true,
+ RawQuery: "thing=value",
+ Fragment: "1234",
+ RawFragment: "1234",
+ },
+ },
+ }
+ for i, tt := range tests {
+ t.Run(strconv.Itoa(i), func(t *testing.T) {
+ gotClone := cloneURL(tt.value)
+ if gotClone == tt.value {
+ t.Errorf("expct clone URL to not be same pointer address")
+ }
+ if tt.value.User != nil {
+ if tt.value.User == gotClone.User {
+ t.Errorf("expct cloned Userinfo to not be same pointer address")
+ }
+ }
+ if !reflect.DeepEqual(gotClone, tt.wantClone) {
+ t.Errorf("cloneURL() = %v, want %v", gotClone, tt.wantClone)
+ }
+ })
+ }
+}
+
+func TestEndpoint_Prune(t *testing.T) {
+ endpoint := Endpoint{}
+
+ endpoint.Add(WeightedAddress{
+ URL: &url.URL{},
+ Expired: time.Now().Add(5 * time.Minute),
+ })
+
+ initial := endpoint.Addresses
+
+ if e, a := false, endpoint.Prune(); e != a {
+ t.Errorf("expect prune %v, got %v", e, a)
+ }
+
+ if e, a := &initial[0], &endpoint.Addresses[0]; e != a {
+ t.Errorf("expect slice address to be same")
+ }
+
+ endpoint.Add(WeightedAddress{
+ URL: &url.URL{},
+ Expired: time.Now().Add(5 * -time.Minute),
+ })
+
+ initial = endpoint.Addresses
+
+ if e, a := true, endpoint.Prune(); e != a {
+ t.Errorf("expect prune %v, got %v", e, a)
+ }
+
+ if e, a := &initial[0], &endpoint.Addresses[0]; e == a {
+ t.Errorf("expect slice address to be different")
+ }
+
+ if e, a := 1, endpoint.Len(); e != a {
+ t.Errorf("expect slice length %v, got %v", e, a)
+ }
+}