Improve handling of concurrent identical queries (#146)

* fix(concurrent-queries): improve handling of identical queries

* feat(transactionRegistry): created, completed, failed states

* Concurrent test suites to verify transactions behaviour. Inmemory transaction registry fixes.

* fix(proxy_test): queue overflow test

* fix(review): remove ErrMissingTransaction. remove max exec time/2 check after awaiting concurrent transaction.

* fix(review): use max execution time as default

* fix(review): deprecation in README. Simplify transaction output.

Author: gontarzpawel

cache/async_cache.go

@@ -4,14 +4,17 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/contentsquare/chproxy/log"
+
 	"github.com/contentsquare/chproxy/clients"
 	"github.com/contentsquare/chproxy/config"
 	"github.com/go-redis/redis/v8"
 )
 
-// AsyncCache is a transactional cache allowing the results from concurrent queries.
-// When query A is equal to query B and A arrives no more than defined graceTime, query A will await for the results of query B for the max time equal to:
-// graceTime - (arrivalB - arrivalA)
+// AsyncCache is a transactional cache enabled to serve the results from concurrent queries.
+// When query A and B are equal, and query B arrives after query A with no more than defined deadline interval [[graceTime]],
+// query B will await for the results of query B for the max time equal to:
+// max_awaiting_time = graceTime - (arrivalB - arrivalA)
 type AsyncCache struct {
 	Cache
 	TransactionRegistry
@@ -29,22 +32,28 @@ func (c *AsyncCache) Close() error {
 	return nil
 }
 
-func (c *AsyncCache) AwaitForConcurrentTransaction(key *Key) bool {
+func (c *AsyncCache) AwaitForConcurrentTransaction(key *Key) (TransactionState, error) {
 	startTime := time.Now()
-
+	seenState := transactionAbsent
 	for {
-		if time.Since(startTime) > c.graceTime {
-			// The entry didn't appear during graceTime.
+		elapsedTime := time.Since(startTime)
+		if elapsedTime > c.graceTime {
+			// The entry didn't appear during deadline.
 			// Let the caller creating it.
-			return false
+			return seenState, nil
+		}
+
+		state, err := c.TransactionRegistry.Status(key)
+
+		if err != nil {
+			return seenState, err
 		}
 
-		ok := c.TransactionRegistry.IsDone(key)
-		if ok {
-			return ok
+		if !state.IsPending() {
+			return state, nil
 		}
 
-		// Wait for graceTime in the hope the entry will appear
+		// Wait for deadline in the hope the entry will appear
 		// in the cache.
 		//
 		// This should protect from thundering herd problem when
@@ -57,11 +66,14 @@ func (c *AsyncCache) AwaitForConcurrentTransaction(key *Key) bool {
 	}
 }
 
-func NewAsyncCache(cfg config.Cache) (*AsyncCache, error) {
+func NewAsyncCache(cfg config.Cache, maxExecutionTime time.Duration) (*AsyncCache, error) {
 	graceTime := time.Duration(cfg.GraceTime)
+	if graceTime > 0 {
+		log.Errorf("[DEPRECATED] detected grace time configuration %s. It will be removed in the new version", graceTime)
+	}
 	if graceTime == 0 {
 		// Default grace time.
-		graceTime = 5 * time.Second
+		graceTime = maxExecutionTime
 	}
 	if graceTime < 0 {
 		// Disable protection from `dogpile effect`.
@@ -71,11 +83,13 @@ func NewAsyncCache(cfg config.Cache) (*AsyncCache, error) {
 	var cache Cache
 	var transaction TransactionRegistry
 	var err error
+	// transaction will be kept until we're sure there's no possible concurrent query running
+	transactionDeadline := 2 * graceTime
 
 	switch cfg.Mode {
 	case "file_system":
 		cache, err = newFilesSystemCache(cfg, graceTime)
-		transaction = newInMemoryTransactionRegistry(graceTime)
+		transaction = newInMemoryTransactionRegistry(transactionDeadline)
 	case "redis":
 		var redisClient redis.UniversalClient
 		redisClient, err = clients.NewRedisClient(cfg.Redis)

cache/async_cache_test.go

@@ -1,6 +1,7 @@
 package cache
 
 import (
+	"github.com/stretchr/testify/assert"
 	"log"
 	"os"
 	"testing"
@@ -15,104 +16,169 @@ const asyncTestDir = "./async-test-data"
 func TestAsyncCache_Cleanup_Of_Expired_Transactions(t *testing.T) {
 	graceTime := 100 * time.Millisecond
 	asyncCache := newAsyncTestCache(t, graceTime)
+	defer func() {
+		asyncCache.Close()
+		os.RemoveAll(asyncTestDir)
+	}()
 
 	key := &Key{
 		Query: []byte("SELECT async cache"),
 	}
-
-	if done := asyncCache.IsDone(key); !done {
+	status, err := asyncCache.Status(key)
+	assert.NoError(t, err)
+	if !status.IsAbsent() {
 		t.Fatalf("unexpected behaviour: transaction isnt done while it wasnt even started")
 	}
 
-	if err := asyncCache.Register(key); err != nil {
+	if err := asyncCache.Create(key); err != nil {
 		t.Fatalf("unexpected error: %s failed to register transaction", err)
 	}
 
-	if done := asyncCache.IsDone(key); done {
+	status, err = asyncCache.Status(key)
+	assert.NoError(t, err)
+	if !status.IsPending() {
 		t.Fatalf("unexpected behaviour: transaction isnt finished")
 	}
 
 	time.Sleep(graceTime * 2)
 
-	if done := asyncCache.IsDone(key); !done {
+	status, err = asyncCache.Status(key)
+	assert.NoError(t, err)
+	if status.IsPending() {
 		t.Fatalf("unexpected behaviour: transaction grace time elapsed and yet it was still pending")
 	}
-
-	asyncCache.Close()
-	os.RemoveAll(asyncTestDir)
 }
 
 func TestAsyncCache_AwaitForConcurrentTransaction_GraceTimeWithoutTransactionCompletion(t *testing.T) {
-	graceTime := 300 * time.Millisecond
+	graceTime := 100 * time.Millisecond
 	asyncCache := newAsyncTestCache(t, graceTime)
 
+	defer func() {
+		asyncCache.Close()
+		os.RemoveAll(asyncTestDir)
+	}()
+
 	key := &Key{
 		Query: []byte("SELECT async cache AwaitForConcurrentTransaction"),
 	}
 
-	if done := asyncCache.IsDone(key); !done {
+	status, err := asyncCache.Status(key)
+	assert.NoError(t, err)
+	if !status.IsAbsent() {
 		t.Fatalf("unexpected behaviour: transaction isnt done while it wasnt even started")
 	}
 
-	if err := asyncCache.Register(key); err != nil {
+	if err := asyncCache.Create(key); err != nil {
 		t.Fatalf("unexpected error: %s failed to register transaction", err)
 	}
 
-	if done := asyncCache.IsDone(key); done {
+	status, err = asyncCache.Status(key)
+	assert.NoError(t, err)
+	if !status.IsPending() {
 		t.Fatalf("unexpected behaviour: transaction isnt finished")
 	}
 
 	startTime := time.Now()
-	done := asyncCache.AwaitForConcurrentTransaction(key)
+	_, err = asyncCache.AwaitForConcurrentTransaction(key)
+	assert.NoError(t, err)
 	elapsedTime := time.Since(startTime)
 
 	// in order to let the cleaner swipe the transaction
-	time.Sleep(100 * time.Millisecond)
-	if done == asyncCache.IsDone(key) && done {
+	time.Sleep(150 * time.Millisecond)
+	status, err = asyncCache.Status(key)
+	assert.NoError(t, err)
+	if !status.IsAbsent() {
 		t.Fatalf("unexpected behaviour: transaction awaiting time elapsed %s", elapsedTime.String())
 	}
-
-	asyncCache.Close()
-	os.RemoveAll(asyncTestDir)
 }
 
 func TestAsyncCache_AwaitForConcurrentTransaction_TransactionCompletedWhileAwaiting(t *testing.T) {
 	graceTime := 300 * time.Millisecond
 	asyncCache := newAsyncTestCache(t, graceTime)
 
+	defer func() {
+		asyncCache.Close()
+		os.RemoveAll(asyncTestDir)
+	}()
 	key := &Key{
 		Query: []byte("SELECT async cache AwaitForConcurrentTransactionCompleted"),
 	}
 
-	if err := asyncCache.Register(key); err != nil {
+	if err := asyncCache.Create(key); err != nil {
 		t.Fatalf("unexpected error: %s failed to register transaction", err)
 	}
 
 	errs := make(chan error)
 	go func() {
 		time.Sleep(graceTime / 2)
-		if err := asyncCache.Unregister(key); err != nil {
+		if err := asyncCache.Complete(key); err != nil {
 			errs <- err
 		} else {
 			errs <- nil
 		}
 	}()
 
 	startTime := time.Now()
-	done := asyncCache.AwaitForConcurrentTransaction(key)
+	transactionState, err := asyncCache.AwaitForConcurrentTransaction(key)
+	if err != nil {
+		t.Fatalf("unexpected error: %s failed to unregister transaction", err)
+	}
+
 	elapsedTime := time.Since(startTime)
 
-	err := <-errs
+	err = <-errs
 	if err != nil {
 		t.Fatalf("unexpected error: %s failed to unregister transaction", err)
 	}
 
-	if done != asyncCache.IsDone(key) || !done || elapsedTime >= graceTime {
+	if !transactionState.IsCompleted() || elapsedTime >= graceTime {
 		t.Fatalf("unexpected behaviour: transaction awaiting time elapsed %s", elapsedTime.String())
 	}
+}
+
+func TestAsyncCache_AwaitForConcurrentTransaction_TransactionFailedWhileAwaiting(t *testing.T) {
+	graceTime := 300 * time.Millisecond
+	asyncCache := newAsyncTestCache(t, graceTime)
+
+	defer func() {
+		asyncCache.Close()
+		os.RemoveAll(asyncTestDir)
+	}()
+
+	key := &Key{
+		Query: []byte("SELECT async cache AwaitForConcurrentTransactionCompleted"),
+	}
+
+	if err := asyncCache.Create(key); err != nil {
+		t.Fatalf("unexpected error: %s failed to register transaction", err)
+	}
+
+	errs := make(chan error)
+	go func() {
+		time.Sleep(graceTime / 2)
+		if err := asyncCache.Fail(key); err != nil {
+			errs <- err
+		} else {
+			errs <- nil
+		}
+	}()
+
+	startTime := time.Now()
+	transactionState, err := asyncCache.AwaitForConcurrentTransaction(key)
+	if err != nil {
+		t.Fatalf("unexpected error: %s failed to unregister transaction", err)
+	}
+
+	elapsedTime := time.Since(startTime)
+
+	err = <-errs
+	if err != nil {
+		t.Fatalf("unexpected error: %s failed to unregister transaction", err)
+	}
 
-	asyncCache.Close()
-	os.RemoveAll(asyncTestDir)
+	if !transactionState.IsFailed() || elapsedTime >= graceTime {
+		t.Fatalf("unexpected behaviour: transaction awaiting time elapsed %s", elapsedTime.String())
+	}
 }
 
 func newAsyncTestCache(t *testing.T, graceTime time.Duration) *AsyncCache {
@@ -152,7 +218,7 @@ func TestAsyncCache_FilesystemCache_instantiation(t *testing.T) {
 	if err := os.RemoveAll(testDirAsync); err != nil {
 		log.Fatalf("cannot remove %q: %s", testDirAsync, err)
 	}
-	_, err := NewAsyncCache(fileSystemCfg)
+	_, err := NewAsyncCache(fileSystemCfg, 1*time.Second)
 	if err != nil {
 		t.Fatalf("could not instanciate filsystem async cache because of the following error: %s", err)
 	}
@@ -168,7 +234,7 @@ func TestAsyncCache_FilesystemCache_wrong_instantiation(t *testing.T) {
 		},
 		Expire: config.Duration(time.Minute),
 	}
-	_, err := NewAsyncCache(fileSystemCfg)
+	_, err := NewAsyncCache(fileSystemCfg, 1*time.Second)
 	if err == nil {
 		t.Fatalf("the instanciate of filsystem async cache should have crashed")
 	}
@@ -185,7 +251,7 @@ func TestAsyncCache_RedisCache_instantiation(t *testing.T) {
 		Expire: config.Duration(cacheTTL),
 	}
 
-	_, err := NewAsyncCache(redisCfg)
+	_, err := NewAsyncCache(redisCfg, 1*time.Second)
 	if err != nil {
 		t.Fatalf("could not instanciate redis async cache because of the following error: %s", err)
 	}
@@ -201,21 +267,21 @@ func TestAsyncCache_RedisCache_wrong_instantiation(t *testing.T) {
 		},
 	}
 
-	_, err := NewAsyncCache(redisCfg)
+	_, err := NewAsyncCache(redisCfg, 1*time.Second)
 	if err == nil {
 		t.Fatalf("the redis instanciation should have crashed")
 	}
 }
 
-func TestAsyncCache_Unkown_instantiation(t *testing.T) {
+func TestAsyncCache_Unknown_instantiation(t *testing.T) {
 	var redisCfg = config.Cache{
 		Name:   "test",
 		Mode:   "Unkown Mode",
 		Redis:  config.RedisCacheConfig{},
 		Expire: config.Duration(cacheTTL),
 	}
 
-	_, err := NewAsyncCache(redisCfg)
+	_, err := NewAsyncCache(redisCfg, 1*time.Second)
 	if err == nil {
 		t.Fatalf("The instanciation should have crash")
 	}

cache/filesystem_cache.go

@@ -115,7 +115,7 @@ func (f *fileSystemCache) Get(key *Key) (*CachedData, error) {
 			return nil, ErrMissing
 		}
 		// Serve expired file in the hope it will be substituted
-		// with the fresh file during graceTime.
+		// with the fresh file during deadline.
 	}
 
 	b, err := ioutil.ReadAll(file)
@@ -242,7 +242,7 @@ func (f *fileSystemCache) clean() {
 
 	log.Debugf("cache %q: start cleaning dir %q", f.Name(), f.dir)
 
-	// Remove cached files after a graceTime from their expiration,
+	// Remove cached files after a deadline from their expiration,
 	// so they may be served until they are substituted with fresh files.
 	expire := f.expire + f.grace
 

cache/redis_cache.go

@@ -97,7 +97,7 @@ func (r *redisCache) Get(key *Key) (*CachedData, error) {
 	val, err := r.client.Get(ctx, key.String()).Result()
 
 	// if key not found in cache
-	if errors.Is(err, redis.Nil) || val == pendingTransactionVal {
+	if errors.Is(err, redis.Nil) {
 		return nil, ErrMissing
 	}