topdown: Update cache interface with Clone method

Concurrent evaluation of the http.send builtin for the
same object can sometimes result in the HTTP headers
map being concurrently accessed. This can happen for
example when a key already present in the inter-query
cache needs to be revalidated and multiple routines
may access the HTTP headers at the same time resulting
in a race.

This change adds a new Clone method to cache interface.
The idea is to give each routine its own copy of the cached object
which would mean it has a copy of the headers map and
thus should be able to avoid any sync issues.

Signed-off-by: Ashutosh Narkar <anarkar4387@gmail.com>

topdown/cache/cache.go

@@ -7,11 +7,10 @@ package cache
 
 import (
 	"container/list"
+	"sync"
 
 	"github.com/open-policy-agent/opa/ast"
 
-	"sync"
-
 	"github.com/open-policy-agent/opa/util"
 )
 
@@ -62,6 +61,7 @@ func (c *Config) validateAndInjectDefaults() error {
 // InterQueryCacheValue defines the interface for the data that the inter-query cache holds.
 type InterQueryCacheValue interface {
 	SizeInBytes() int64
+	Clone() (InterQueryCacheValue, error)
 }
 
 // InterQueryCache defines the interface for the inter-query cache.
@@ -70,6 +70,7 @@ type InterQueryCache interface {
 	Insert(key ast.Value, value InterQueryCacheValue) int
 	Delete(key ast.Value)
 	UpdateConfig(config *Config)
+	Clone(value InterQueryCacheValue) (InterQueryCacheValue, error)
 }
 
 // NewInterQueryCache returns a new inter-query cache.
@@ -130,6 +131,12 @@ func (c *cache) UpdateConfig(config *Config) {
 	c.config = config
 }
 
+func (c *cache) Clone(value InterQueryCacheValue) (InterQueryCacheValue, error) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.unsafeClone(value)
+}
+
 func (c *cache) unsafeInsert(k ast.Value, v InterQueryCacheValue) (dropped int) {
 	size := v.SizeInBytes()
 	limit := c.maxSizeBytes()
@@ -174,6 +181,10 @@ func (c *cache) unsafeDelete(k ast.Value) {
 	c.l.Remove(cacheItem.keyElement)
 }
 
+func (c *cache) unsafeClone(value InterQueryCacheValue) (InterQueryCacheValue, error) {
+	return value.Clone()
+}
+
 func (c *cache) maxSizeBytes() int64 {
 	if c.config == nil {
 		return defaultMaxSizeBytes

topdown/cache/cache_test.go

@@ -211,6 +211,47 @@ func TestConcurrentInsert(t *testing.T) {
 	}
 }
 
+func TestClone(t *testing.T) {
+	in := `{"inter_query_builtin_cache": {"max_size_bytes": 40},}`
+
+	config, err := ParseCachingConfig([]byte(in))
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	cache := NewInterQueryCache(config)
+
+	cacheValue := newInterQueryCacheValue(ast.StringTerm("bar").Value, 20)
+	dropped := cache.Insert(ast.StringTerm("foo").Value, cacheValue)
+	if dropped != 0 {
+		t.Fatal("Expected dropped to be zero")
+	}
+
+	val, found := cache.Get(ast.StringTerm("foo").Value)
+	if !found {
+		t.Fatal("Expected key \"foo\" in cache")
+	}
+
+	dup, err := cache.Clone(val)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	original, ok := val.(*testInterQueryCacheValue)
+	if !ok {
+		t.Fatal("unexpected type")
+	}
+
+	cloned, ok := dup.(*testInterQueryCacheValue)
+	if !ok {
+		t.Fatal("unexpected type")
+	}
+
+	if !reflect.DeepEqual(*original, *cloned) {
+		t.Fatalf("Expected to get %v, but got %v", *original, *cloned)
+	}
+}
+
 func TestDelete(t *testing.T) {
 	config, err := ParseCachingConfig(nil)
 	if err != nil {
@@ -299,3 +340,7 @@ func newInterQueryCacheValue(value ast.Value, size int) *testInterQueryCacheValu
 func (p testInterQueryCacheValue) SizeInBytes() int64 {
 	return int64(p.size)
 }
+
+func (p testInterQueryCacheValue) Clone() (InterQueryCacheValue, error) {
+	return &testInterQueryCacheValue{value: p.value, size: p.size}, nil
+}

topdown/http.go

@@ -803,10 +803,16 @@ func insertErrorIntoHTTPSendCache(bctx BuiltinContext, key ast.Object, err error
 func (c *interQueryCache) checkHTTPSendInterQueryCache() (ast.Value, error) {
 	requestCache := c.bctx.InterQueryBuiltinCache
 
-	value, found := requestCache.Get(c.key)
+	cachedValue, found := requestCache.Get(c.key)
 	if !found {
 		return nil, nil
 	}
+
+	value, cerr := requestCache.Clone(cachedValue)
+	if cerr != nil {
+		return nil, handleHTTPSendErr(c.bctx, cerr)
+	}
+
 	c.bctx.Metrics.Counter(httpSendInterQueryCacheHits).Incr()
 	var cachedRespData *interQueryCacheData
 
@@ -1035,6 +1041,12 @@ func newInterQueryCacheValue(bctx BuiltinContext, resp *http.Response, respBody
 	return &interQueryCacheValue{Data: b}, nil
 }
 
+func (cb interQueryCacheValue) Clone() (cache.InterQueryCacheValue, error) {
+	dup := make([]byte, len(cb.Data))
+	copy(dup, cb.Data)
+	return &interQueryCacheValue{Data: dup}, nil
+}
+
 func (cb interQueryCacheValue) SizeInBytes() int64 {
 	return int64(len(cb.Data))
 }
@@ -1118,6 +1130,18 @@ func (c *interQueryCacheData) SizeInBytes() int64 {
 	return 0
 }
 
+func (c *interQueryCacheData) Clone() (cache.InterQueryCacheValue, error) {
+	dup := make([]byte, len(c.RespBody))
+	copy(dup, c.RespBody)
+
+	return &interQueryCacheData{
+		ExpiresAt:  c.ExpiresAt,
+		RespBody:   dup,
+		Status:     c.Status,
+		StatusCode: c.StatusCode,
+		Headers:    c.Headers.Clone()}, nil
+}
+
 type responseHeaders struct {
 	date         time.Time         // origination date and time of response
 	cacheControl map[string]string // response cache-control header

topdown/http_test.go

@@ -2851,6 +2851,10 @@ func (c *onlyOnceInterQueryCache) Delete(_ ast.Value) {}
 
 func (c *onlyOnceInterQueryCache) UpdateConfig(_ *iCache.Config) {}
 
+func (c *onlyOnceInterQueryCache) Clone(val iCache.InterQueryCacheValue) (iCache.InterQueryCacheValue, error) {
+	return val, nil
+}
+
 func TestInterQueryCacheConcurrentModification(t *testing.T) {
 
 	// create an inter-query cache that'll return a value on first access, but none at subsequent accesses.
@@ -2899,6 +2903,66 @@ func TestInterQueryCacheConcurrentModification(t *testing.T) {
 	}
 }
 
+func TestInterQueryCacheDataClone(t *testing.T) {
+	data := interQueryCacheData{
+		Headers: map[string][]string{
+			"Date": {"Thu, 01 Jan 1970 00:00:00 GMT"},
+		},
+		ExpiresAt:  time.Now().Add(time.Hour),
+		StatusCode: 200,
+		Status:     "200 OK",
+		RespBody:   []byte("foo"),
+	}
+
+	dup, err := data.Clone()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cloned, ok := dup.(*interQueryCacheData)
+	if !ok {
+		t.Fatal("unexpected type")
+	}
+
+	if !reflect.DeepEqual(data, *cloned) {
+		t.Fatalf("Expected to get %v, but got %v", data, *cloned)
+	}
+}
+
+func TestInterQueryCacheValueClone(t *testing.T) {
+
+	cacheData := interQueryCacheData{
+		Headers: map[string][]string{
+			"Date": {"Thu, 01 Jan 1970 00:00:00 GMT"},
+		},
+		ExpiresAt:  time.Now().Add(time.Hour),
+		StatusCode: 200,
+		Status:     "200 OK",
+		RespBody:   []byte("foo"),
+	}
+
+	b, err := json.Marshal(cacheData)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cacheVal := interQueryCacheValue{Data: b}
+
+	dup, err := cacheVal.Clone()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cloned, ok := dup.(*interQueryCacheValue)
+	if !ok {
+		t.Fatal("unexpected type")
+	}
+
+	if !reflect.DeepEqual(cacheVal, *cloned) {
+		t.Fatal("inter-query cache element and its clone are not equal")
+	}
+}
+
 func TestIntraQueryCache_ClientError(t *testing.T) {
 	data := loadSmallTestData()