Don't coalesce anymore uncoalesceable request (#56)

* Don't coalesce anymore uncoalesceable request

* Add uncoalesceable request to the storage layer

* Update LayerStorage behaviour

Author: darkweak

cache/types/layerStorage.go

@@ -8,6 +8,16 @@ type VaryLayerStorage struct {
 	*ristretto.Cache
 }
 
+func InitializeVaryLayerStorage() *VaryLayerStorage {
+	storage, _ := ristretto.NewCache(&ristretto.Config{
+		NumCounters: 1e7,     // number of keys to track frequency of (10M).
+		MaxCost:     1 << 30, // maximum cost of cache (1GB).
+		BufferItems: 64,      // number of keys per Get buffer.
+	})
+
+	return &VaryLayerStorage{Cache: storage}
+}
+
 // Get method returns the varied headers list if exists, empty array then
 func (provider *VaryLayerStorage) Get(key string) []string {
 	val, found := provider.Cache.Get(key)
@@ -24,3 +34,38 @@ func (provider *VaryLayerStorage) Set(key string, headers []string) {
 		panic("Impossible to set value into Ristretto")
 	}
 }
+
+type CoalescingLayerStorage struct {
+	*ristretto.Cache
+}
+
+func InitializeCoalescingLayerStorage() *CoalescingLayerStorage {
+	storage, _ := ristretto.NewCache(&ristretto.Config{
+		NumCounters: 1e7,     // number of keys to track frequency of (10M).
+		MaxCost:     1 << 30, // maximum cost of cache (1GB).
+		BufferItems: 64,      // number of keys per Get buffer.
+	})
+
+	return &CoalescingLayerStorage{Cache: storage}
+}
+
+// Exists method returns if the key should coalesce
+func (provider *CoalescingLayerStorage) Exists(key string) bool {
+	_, found := provider.Cache.Get(key)
+	return !found
+}
+
+// Set method will store the response in Ristretto provider
+func (provider *CoalescingLayerStorage) Set(key string) {
+	isSet := provider.Cache.Set(key, nil, 1)
+	if !isSet {
+		panic("Impossible to set value into Ristretto")
+	}
+}
+
+// Delete method will delete the response in Ristretto provider if exists corresponding to key param
+func (provider *CoalescingLayerStorage) Delete(key string) {
+	go func() {
+		provider.Del(key)
+	}()
+}

cache/types/layerStorage_test.go

@@ -0,0 +1,48 @@
+package types
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/darkweak/souin/errors"
+	"time"
+)
+
+const LAYEREDKEY = "LayeredKey"
+const BYTEKEY = "MyByteKey"
+const NONEXISTENTKEY = "NonexistentKey"
+
+func TestInitializeCoalescingLayerStorage(t *testing.T) {
+	r := InitializeCoalescingLayerStorage()
+
+	if nil == r || nil == r.Cache {
+		errors.GenerateError(t, "Ristretto should be instanciated")
+	}
+}
+
+func TestIShouldBeAbleToReadAndWriteDataInTheLayerStorage(t *testing.T) {
+	store := InitializeCoalescingLayerStorage()
+
+	store.Set(LAYEREDKEY)
+	time.Sleep(1 * time.Second)
+
+	if store.Exists(LAYEREDKEY) {
+		errors.GenerateError(t, fmt.Sprintf("Key %s should exist", LAYEREDKEY))
+	}
+}
+
+func TestLayerStorage_GetRequestInTheLayerStorage(t *testing.T) {
+	store := InitializeCoalescingLayerStorage()
+	if !store.Exists(NONEXISTENTKEY) {
+		errors.GenerateError(t, fmt.Sprintf("Key %s should not exist", NONEXISTENTKEY))
+	}
+}
+
+func TestLayerStorage_DeleteRequestInCache(t *testing.T) {
+	store := InitializeCoalescingLayerStorage()
+	store.Delete(BYTEKEY)
+	time.Sleep(1 * time.Second)
+	if !store.Exists(BYTEKEY) {
+		errors.GenerateError(t, fmt.Sprintf("Key %s should not exist", BYTEKEY))
+	}
+}

cache/types/souin.go

@@ -13,6 +13,7 @@ type TransportInterface interface {
 	SetURL(url configurationtypes.URL)
 	UpdateCacheEventually(req *http.Request) (resp *http.Response, err error)
 	GetVaryLayerStorage() *VaryLayerStorage
+	GetCoalescingLayerStorage() *CoalescingLayerStorage
 }
 
 // Transport is an implementation of http.RoundTripper that will return values from a cache
@@ -21,11 +22,12 @@ type TransportInterface interface {
 type Transport struct {
 	// The RoundTripper interface actually used to make requests
 	// If nil, http.DefaultTransport is used
-	Transport           http.RoundTripper
-	Provider            AbstractProviderInterface
-	ConfigurationURL    configurationtypes.URL
-	MarkCachedResponses bool
-	VaryLayerStorage    *VaryLayerStorage
+	Transport              http.RoundTripper
+	Provider               AbstractProviderInterface
+	ConfigurationURL       configurationtypes.URL
+	MarkCachedResponses    bool
+	VaryLayerStorage       *VaryLayerStorage
+	CoalescingLayerStorage *CoalescingLayerStorage
 }
 
 // RetrieverResponsePropertiesInterface interface

plugins/base.go

@@ -20,13 +20,17 @@ func DefaultSouinPluginCallback(
 	nextMiddleware func(w http.ResponseWriter, r *http.Request) error,
 ) {
 	responses := make(chan types.ReverseResponse)
+	coalesceable := make(chan bool)
 
 	go func() {
 		cacheKey := rfc.GetCacheKey(req)
 		varied := retriever.GetTransport().GetVaryLayerStorage().Get(cacheKey)
 		if len(varied) != 0 {
 			cacheKey = rfc.GetVariedCacheKey(req, varied)
 		}
+		go func() {
+			coalesceable <- retriever.GetTransport().GetCoalescingLayerStorage().Exists(cacheKey)
+		}()
 		if http.MethodGet == req.Method {
 			r, _ := rfc.CachedResponse(
 				retriever.GetProvider(),
@@ -56,7 +60,12 @@ func DefaultSouinPluginCallback(
 	}
 
 	close(responses)
-	rc.Temporise(req, res, nextMiddleware)
+	if <-coalesceable {
+		rc.Temporise(req, res, nextMiddleware)
+	} else {
+		_ = nextMiddleware(res, req)
+	}
+	close(coalesceable)
 }
 
 // DefaultSouinPluginInitializerFromConfiguration is the default initialization for plugins

rfc/bridge.go

@@ -47,43 +47,55 @@ func (t *VaryTransport) BaseRoundTrip(req *http.Request, shouldReUpdate bool) (s
 			cachedResp = cr.Response
 		}
 	} else {
+		go func() {
+			t.CoalescingLayerStorage.Set(cacheKey)
+		}()
 		t.Provider.Delete(cacheKey)
 	}
 
 	return cacheKey, cacheable, cachedResp
 }
 
-// UpdateCacheEventually will handle Request and update the previous one in the cache provider
-func (t *VaryTransport) UpdateCacheEventually(req *http.Request) (resp *http.Response, err error) {
-	cacheKey, cacheable, cachedResp := t.BaseRoundTrip(req, false)
+func commonVaryMatchesVerification(cachedResp *http.Response, req *http.Request) *http.Response {
+	if varyMatches(cachedResp, req) {
+		// Can only use cached value if the new request doesn't Vary significantly
+		freshness := getFreshness(cachedResp.Header, req.Header)
+		if freshness == fresh {
+			return cachedResp
+		}
 
-	if cacheable && cachedResp != nil {
-		if varyMatches(cachedResp, req) {
-			// Can only use cached value if the new request doesn't Vary significantly
-			freshness := getFreshness(cachedResp.Header, req.Header)
-			if freshness == fresh {
-				return cachedResp, nil
+		if freshness == stale {
+			var req2 *http.Request
+			// Add validators if caller hasn't already done so
+			etag := cachedResp.Header.Get("etag")
+			if etag != "" && req.Header.Get("etag") == "" {
+				req2 = cloneRequest(req)
+				req2.Header.Set("if-none-match", etag)
 			}
-
-			if freshness == stale {
-				var req2 *http.Request
-				// Add validators if caller hasn't already done so
-				etag := cachedResp.Header.Get("etag")
-				if etag != "" && req.Header.Get("etag") == "" {
+			lastModified := cachedResp.Header.Get("last-modified")
+			if lastModified != "" && req.Header.Get("last-modified") == "" {
+				if req2 == nil {
 					req2 = cloneRequest(req)
-					req2.Header.Set("if-none-match", etag)
-				}
-				lastModified := cachedResp.Header.Get("last-modified")
-				if lastModified != "" && req.Header.Get("last-modified") == "" {
-					if req2 == nil {
-						req2 = cloneRequest(req)
-					}
-					req2.Header.Set("if-modified-since", lastModified)
-				}
-				if req2 != nil {
-					req = req2
 				}
+				req2.Header.Set("if-modified-since", lastModified)
 			}
+			if req2 != nil {
+				req = req2
+			}
+		}
+	}
+
+	return nil
+}
+
+// UpdateCacheEventually will handle Request and update the previous one in the cache provider
+func (t *VaryTransport) UpdateCacheEventually(req *http.Request) (resp *http.Response, err error) {
+	cacheKey, cacheable, cachedResp := t.BaseRoundTrip(req, false)
+
+	if cacheable && cachedResp != nil {
+		r := commonVaryMatchesVerification(cachedResp, req)
+		if r != nil {
+			return r, nil
 		}
 	} else {
 		reqCacheControl := parseCacheControl(req.Header)
@@ -126,32 +138,9 @@ func (t *VaryTransport) RoundTrip(req *http.Request) (resp *http.Response, err e
 			cachedResp.Header.Set(XFromCache, "1")
 		}
 
-		if varyMatches(cachedResp, req) {
-			// Can only use cached value if the new request doesn't Vary significantly
-			freshness := getFreshness(cachedResp.Header, req.Header)
-			if freshness == fresh {
-				return cachedResp, nil
-			}
-
-			if freshness == stale {
-				var req2 *http.Request
-				// Add validators if caller hasn't already done so
-				etag := cachedResp.Header.Get("etag")
-				if etag != "" && req.Header.Get("etag") == "" {
-					req2 = cloneRequest(req)
-					req2.Header.Set("if-none-match", etag)
-				}
-				lastModified := cachedResp.Header.Get("last-modified")
-				if lastModified != "" && req.Header.Get("last-modified") == "" {
-					if req2 == nil {
-						req2 = cloneRequest(req)
-					}
-					req2.Header.Set("if-modified-since", lastModified)
-				}
-				if req2 != nil {
-					req = req2
-				}
-			}
+		r := commonVaryMatchesVerification(cachedResp, req)
+		if r != nil {
+			return r, nil
 		}
 
 		resp, err = transport.RoundTrip(req)
@@ -188,6 +177,9 @@ func (t *VaryTransport) RoundTrip(req *http.Request) (resp *http.Response, err e
 	}
 	resp, err = transport.RoundTrip(req)
 	if !(cacheable && validateVary(req, resp, cacheKey, t)) {
+		go func() {
+			t.CoalescingLayerStorage.Set(cacheKey)
+		}()
 		t.Provider.Delete(cacheKey)
 	}
 	return resp, nil

rfc/transport.go

@@ -3,7 +3,6 @@ package rfc
 import (
 	"github.com/darkweak/souin/cache/types"
 	"github.com/darkweak/souin/configurationtypes"
-	"github.com/dgraph-io/ristretto"
 	"github.com/pquerna/cachecontrol"
 	"net/http"
 	"net/http/httputil"
@@ -23,13 +22,12 @@ func IsVaryCacheable(req *http.Request) bool {
 // NewTransport returns a new Transport with the
 // provided Cache implementation and MarkCachedResponses set to true
 func NewTransport(p types.AbstractProviderInterface) *VaryTransport {
-	storage, _ := ristretto.NewCache(&ristretto.Config{
-		NumCounters: 1e7,     // number of keys to track frequency of (10M).
-		MaxCost:     1 << 30, // maximum cost of cache (1GB).
-		BufferItems: 64,      // number of keys per Get buffer.
-		OnEvict: func(key, conflict uint64, value interface{}, cost int64) {},
-	})
-	return &VaryTransport{Provider: p, VaryLayerStorage: &types.VaryLayerStorage{Cache: storage}, MarkCachedResponses: true}
+	return &VaryTransport{
+		Provider:               p,
+		VaryLayerStorage:       types.InitializeVaryLayerStorage(),
+		CoalescingLayerStorage: types.InitializeCoalescingLayerStorage(),
+		MarkCachedResponses:    true,
+	}
 }
 
 // GetProvider returns the associated provider
@@ -46,6 +44,10 @@ func (t *VaryTransport) GetVaryLayerStorage() *types.VaryLayerStorage {
 	return t.VaryLayerStorage
 }
 
+func (t *VaryTransport) GetCoalescingLayerStorage() *types.CoalescingLayerStorage {
+	return t.CoalescingLayerStorage
+}
+
 // SetCache set the cache
 func (t *VaryTransport) SetCache(key string, resp *http.Response, req *http.Request) {
 	r, _, _ := cachecontrol.CachableResponse(req, resp, cachecontrol.Options{})

rfc/vary.go

@@ -11,7 +11,7 @@ import (
 func varyMatches(cachedResp *http.Response, req *http.Request) bool {
 	for _, header := range headerAllCommaSepValues(cachedResp.Header, "vary") {
 		header = http.CanonicalHeaderKey(header)
-		if header != "" && req.Header.Get(header) != cachedResp.Header.Get("X-Varied-"+header) {
+		if header == "" || req.Header.Get(header) == "" {
 			return false
 		}
 	}
@@ -26,14 +26,6 @@ func validateVary(req *http.Request, resp *http.Response, key string, t *VaryTra
 			go func() {
 				t.VaryLayerStorage.Set(key, variedHeaders)
 			}()
-			for _, varyKey := range variedHeaders {
-				varyKey = http.CanonicalHeaderKey(varyKey)
-				fakeHeader := "X-Varied-" + varyKey
-				reqValue := req.Header.Get(varyKey)
-				if reqValue != "" {
-					resp.Header.Set(fakeHeader, reqValue)
-				}
-			}
 			cacheKey = GetVariedCacheKey(req, variedHeaders)
 		}
 		switch req.Method {
@@ -45,6 +37,9 @@ func validateVary(req *http.Request, resp *http.Response, key string, t *VaryTra
 					resp := *resp
 					resp.Body = ioutil.NopCloser(r)
 					t.SetCache(cacheKey, &resp, req)
+					go func() {
+						t.CoalescingLayerStorage.Delete(cacheKey)
+					}()
 				},
 			}
 		}

rfc/vary_test.go

@@ -25,23 +25,21 @@ func TestVaryMatches(t *testing.T) {
 	}
 
 	header := "Cache"
-	r.Header.Set("Vary", header)
 	r.Header.Set(header, "same")
 	res.Header.Set("vary", header)
-	res.Header.Set("X-Varied-"+header, "same")
 
 	if !varyMatches(res, r) {
-		errors.GenerateError(t, "Vary match should return true if Response contains X-Varied-* header is the same than * in Request header")
+		errors.GenerateError(t, "Vary match should return true if Response contains a vary header that is not null in the request")
 	}
 
 	if !validateVary(r, res, GetCacheKey(r), tr) {
 		errors.GenerateError(t, fmt.Sprintf("It contains valid vary headers in the Response. It should validate it, %v given", res.Header))
 	}
 
-	res.Header.Set("X-Varied-"+header, "different")
+	r.Header.Set(header, "")
 
 	if varyMatches(res, r) {
-		errors.GenerateError(t, "Vary match should return false if Response contains X-Varied-* header different than * in Request header")
+		errors.GenerateError(t, "Vary match should return false if Response contains a vary header that is empty in the request")
 	}
 
 	if !validateVary(r, res, GetCacheKey(r), tr) {