Repository: google/exposure-notifications-server
Pull request #557 (2020-06-05 19:32:40)

Factor out cache implementation to standalone package.
* Create write thru cache implementation
* Move database provider to it
* First part of #555

Follow on PR will bring this to other parts of the codebase that use caching.

google-oss-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/557#" title="Author self-approved">mikehelmick</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=google%2Fexposure-notifications-server).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/google/exposure-notifications-server/blob/master/OWNERS)~~ [mikehelmick]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
google-oss-robot: @mikehelmick: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-en-server-release-unit | e289a0f9d0da415c50d82255dfa0979e8257586a | [link](https://oss-prow.knative.dev/view/gcs/oss-prow/pr-logs/pull/google_exposure-notifications-server/557/pull-en-server-release-unit/1268999705958813696) | `/test pull-en-server-release-unit`

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
google-oss-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/557#" title="Author self-approved">mikehelmick</a>*, *<a href="https://github.com/google/exposure-notifications-server/pull/557#pullrequestreview-425605604" title="LGTM">sethvargo</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=google%2Fexposure-notifications-server).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/google/exposure-notifications-server/blob/master/OWNERS)~~ [mikehelmick,sethvargo]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/internal/authorizedapp/database_provider.go b/internal/authorizedapp/database_provider.go
index 76e50e0..185ed1d 100644
--- a/internal/authorizedapp/database_provider.go
+++ b/internal/authorizedapp/database_provider.go
@@ -17,12 +17,13 @@ package authorizedapp
 import (
 	"context"
 	"fmt"
+	"log"
 	"strings"
-	"sync"
 	"time"
 
 	authorizedappdb "github.com/google/exposure-notifications-server/internal/authorizedapp/database"
 	"github.com/google/exposure-notifications-server/internal/authorizedapp/model"
+	"github.com/google/exposure-notifications-server/internal/cache"
 	"github.com/google/exposure-notifications-server/internal/database"
 
 	"github.com/google/exposure-notifications-server/internal/logging"
@@ -39,13 +40,7 @@ type DatabaseProvider struct {
 	secretManager secrets.SecretManager
 	cacheDuration time.Duration
 
-	cache     map[string]*cacheItem
-	cacheLock sync.RWMutex
-}
-
-type cacheItem struct {
-	value    *model.AuthorizedApp
-	cachedAt time.Time
+	cache *cache.Cache
 }
 
 // DatabaseProviderOption is used as input to the database provider.
@@ -61,10 +56,14 @@ func WithSecretManager(sm secrets.SecretManager) DatabaseProviderOption {
 
 // NewDatabaseProvider creates a new Provider that reads from a database.
 func NewDatabaseProvider(ctx context.Context, db *database.DB, config *Config, opts ...DatabaseProviderOption) (Provider, error) {
+	cache, err := cache.New(config.CacheDuration)
+	if err != nil {
+		return nil, fmt.Errorf("cache.New: %w", err)
+	}
 	provider := &DatabaseProvider{
 		database:      db,
 		cacheDuration: config.CacheDuration,
-		cache:         make(map[string]*cacheItem),
+		cache:         cache,
 	}
 
 	// Apply options.
@@ -75,25 +74,6 @@ func NewDatabaseProvider(ctx context.Context, db *database.DB, config *Config, o
 	return provider, nil
 }
 
-// checkCache checks the local cache within a read lock.
-// The bool on return is true if there was a hit (And an error is a valid hit)
-// or false if there was a miss (or expiry) and the data source should be queried again.
-func (p *DatabaseProvider) checkCache(name string) (*model.AuthorizedApp, bool, error) {
-	// Acquire a read lock first, which allows concurrent readers, to check if
-	// there's an item in the cache.
-	p.cacheLock.RLock()
-	defer p.cacheLock.RUnlock()
-
-	item, ok := p.cache[name]
-	if ok && time.Since(item.cachedAt) <= p.cacheDuration {
-		if item.value == nil {
-			return nil, true, ErrAppNotFound
-		}
-		return item.value, true, nil
-	}
-	return nil, false, nil
-}
-
 // AppConfig returns the config for the given app package name.
 func (p *DatabaseProvider) AppConfig(ctx context.Context, name string) (*model.AuthorizedApp, error) {
 	logger := logging.FromContext(ctx)
@@ -102,42 +82,30 @@ func (p *DatabaseProvider) AppConfig(ctx context.Context, name string) (*model.A
 	// cacher does not. To maximize cache hits, convert to lowercase.
 	name = strings.ToLower(name)
 
-	data, cacheHit, error := p.checkCache(name)
-	if cacheHit {
-		return data, error
-	}
-
-	// Acquire a more aggressive lock now because we're about to mutate. However,
-	// it's possible that a concurrent routine has already mutated between our
-	// read and write locks, so we have to check again.
-	p.cacheLock.Lock()
-	defer p.cacheLock.Unlock()
-	item, ok := p.cache[name]
-	if ok && time.Since(item.cachedAt) <= p.cacheDuration {
-		if item.value == nil {
-			return nil, ErrAppNotFound
+	lookup := func() (interface{}, error) {
+		// Load config.
+		config, err := p.loadAuthorizedAppFromDatabase(ctx, name)
+		if err != nil {
+			return nil, fmt.Errorf("authorizedapp: %w", err)
 		}
-		return item.value, nil
+		logger.Infof("authorizedapp: loaded %v, caching for %s", name, p.cacheDuration)
+		return config, nil
 	}
+	cached, err := p.cache.WriteThruLookup(name, lookup)
 
-	// Load config.
-	config, err := p.loadAuthorizedAppFromDatabase(ctx, name)
+	// Indicates an error on the write thru lookup.
 	if err != nil {
-		return nil, fmt.Errorf("authorizedapp: %w", err)
-	}
-
-	// Cache configs.
-	logger.Infof("authorizedapp: loaded %v, caching for %s", name, p.cacheDuration)
-	p.cache[name] = &cacheItem{
-		value:    config,
-		cachedAt: time.Now(),
+		return nil, err
 	}
 
 	// Handle not found.
+	config := cached.(*model.AuthorizedApp)
 	if config == nil {
 		return nil, ErrAppNotFound
 	}
 
+	log.Printf("AppConfig: %+v %v", config, err)
+
 	// Returned config.
 	return config, nil
 }
diff --git a/internal/cache/cache.go b/internal/cache/cache.go
new file mode 100644
index 0000000..976f951
--- /dev/null
+++ b/internal/cache/cache.go
@@ -0,0 +1,141 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package cache implements an inmemory cache for any interface{} object.
+package cache
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+var ErrInvalidDuration = errors.New("expireAfter duration cannot be negative")
+
+type Func func() (interface{}, error)
+
+type Cache struct {
+	data        map[string]item
+	expireAfter time.Duration
+	mu          sync.RWMutex
+}
+
+type item struct {
+	object    interface{}
+	expiresAt int64
+}
+
+func (i *item) expired() bool {
+	return i.expiresAt < time.Now().UnixNano()
+}
+
+// New creates a new in memory cache.
+func New(expireAfter time.Duration) (*Cache, error) {
+	if expireAfter < 0 {
+		return nil, ErrInvalidDuration
+	}
+
+	return &Cache{
+		data:        make(map[string]item),
+		expireAfter: expireAfter,
+	}, nil
+}
+
+// Removes an item by name and expiry time when the purge was scheduled.
+// If there is a race, and the item has been refreshed, it will not be purged.
+func (c *Cache) purgeExpired(name string, expectedExpiryTime int64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if item, ok := c.data[name]; ok && item.expiresAt == expectedExpiryTime {
+		// found, and the expiry time is still the same as when the purge was requested.
+		delete(c.data, name)
+	}
+}
+
+// Size returns the number of items in the cache.
+func (c *Cache) Size() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return len(c.data)
+}
+
+// WriteThruLookup checks the cache for the value associated with name,
+// and if not found or expired, invokes the provided primaryLookup function
+// to local the value.
+func (c *Cache) WriteThruLookup(name string, primaryLookup Func) (interface{}, error) {
+	// This call takes a read lock.
+	val, hit := c.Lookup(name)
+	if hit {
+		return val, nil
+	}
+
+	// Escalate the lock to a RW lock.
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	// double check that the value hasn't been set by another goroutine.
+	if val, hit := c.data[name]; hit && !val.expired() {
+		return val, nil
+	}
+	// Either a miss, or hit w/ expired value.
+
+	// Value does indeed need to be refreshed. Used the provided fucntion.
+	newData, err := primaryLookup()
+	if err != nil {
+		return nil, err
+	}
+
+	// save the newData in the cache. newData may be nil, if that's what the WriteThruFunction provided.
+	c.data[name] = item{
+		object:    newData,
+		expiresAt: time.Now().Add(c.expireAfter).UnixNano(),
+	}
+	return newData, nil
+
+}
+
+// Lookup checks the cache for a non-expired object by the supplied key name.
+// The bool return informs the caller if there was a cache hit or not.
+// A return of nil, true means that nil is in the cache.
+// Where nil, false indicates a cache miss or that the value is expired and should
+// be refreshed.
+func (c *Cache) Lookup(name string) (interface{}, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if item, ok := c.data[name]; ok && item.expired() {
+		// Cache hit, but expired. The removal from the cache is deferred.
+		go c.purgeExpired(name, item.expiresAt)
+		return nil, false
+	} else if ok {
+		// Cache hit, not expired.
+		return item.object, true
+	}
+	// Cache miss.
+	return nil, false
+}
+
+// Set saves the current value of an object in the cache, with the supplied
+// durintion until the object expires.
+func (c *Cache) Set(name string, object interface{}) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.data[name] = item{
+		object:    object,
+		expiresAt: time.Now().Add(c.expireAfter).UnixNano(),
+	}
+
+	return nil
+}


Test patch:
diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go
new file mode 100644
index 0000000..9949e0f
--- /dev/null
+++ b/internal/cache/cache_test.go
@@ -0,0 +1,179 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package cache implements an inmemory cache for any interface{} object.
+package cache
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+type order struct {
+	Burgers int
+	Fries   int
+}
+
+func checkSize(t *testing.T, c *Cache, want int) {
+	t.Helper()
+
+	if got := c.Size(); got != want {
+		t.Errorf("wrong size want: %v, got: %v", want, got)
+	}
+}
+
+func TestCache(t *testing.T) {
+	duration := time.Millisecond * 500
+	cache, err := New(duration)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	checkSize(t, cache, 0)
+
+	if err := cache.Set("foo", &order{2, 1}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	time.Sleep(duration)
+	if got, hit := cache.Lookup("foo"); got != nil || hit {
+		t.Fatalf("key did not expire as expected")
+	}
+
+	if got, hit := cache.Lookup("bar"); got != nil || hit {
+		t.Fatalf("got key that was never inserted")
+	}
+
+	want := &order{42, 37}
+	if err := cache.Set("foo", want); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got, hit := cache.Lookup("foo"); got == nil || !hit {
+		t.Fatalf("lookup failed want: %v, got %v", want, got)
+	} else {
+		if diff := cmp.Diff(want, got); diff != "" {
+			t.Fatalf("mismatch (-want, +got):\n%s", diff)
+		}
+	}
+	time.Sleep(duration * 2)
+	if got, hit := cache.Lookup("foo"); got != nil || hit {
+		t.Fatalf("expected key to expire, but still available")
+	}
+	// potential race, yeild CPU so that the purge go routine has a chance to run.
+	time.Sleep(duration)
+	checkSize(t, cache, 0)
+}
+
+func TestWriteThruCache(t *testing.T) {
+	cache, err := New(time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	lookupCount := 0
+	want := &order{12, 34}
+	lookerUpper := func() (interface{}, error) {
+		lookupCount++
+		return want, nil
+	}
+
+	for i := 0; i < 2; i++ {
+		got, err := cache.WriteThruLookup("foo", lookerUpper)
+		if err != nil {
+			t.Fatalf("unexpected error on WriteThruLookup: %v", err)
+		}
+		if diff := cmp.Diff(want, got); diff != "" {
+			t.Fatalf("mismatch (-want, +got):\n%s", diff)
+		}
+	}
+
+	if lookupCount != 1 {
+		t.Fatalf("incorrect lookup count, want: 1, got: %v", lookupCount)
+	}
+}
+
+func TestWriteThruError(t *testing.T) {
+	cache, err := New(time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	lookerUpper := func() (interface{}, error) {
+		return nil, fmt.Errorf("nope")
+	}
+
+	got, err := cache.WriteThruLookup("foo", lookerUpper)
+	if err == nil {
+		t.Fatalf("expected error, got nil")
+	}
+	if err.Error() != "nope" {
+		t.Errorf("incorrect error, want: `nope` got: %v", err.Error())
+	}
+	if got != nil {
+		t.Errorf("unexpecetd cached item, want: nil, got %v", got)
+	}
+}
+
+func TestInvalidDuration(t *testing.T) {
+	if _, err := New(-1 * time.Second); err == nil {
+		t.Fatal("expecterd error, got nil")
+	} else if strings.Contains(err.Error(), "duration cannot be nagative") {
+		t.Fatalf("wrong error: want: `duration cannot be negative` got: %v", err.Error())
+	}
+}
+
+func TestConcurrentReaders(t *testing.T) {
+	cache, err := New(time.Second * 5)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	lookupCount := 0
+	want := &order{12, 34}
+	lookerUpper := func() (interface{}, error) {
+		lookupCount++
+		return want, nil
+	}
+
+	done := make(chan bool, 10)
+	for i := 0; i < 10; i++ {
+		ver := i
+		go func() {
+			got, err := cache.WriteThruLookup("foo", lookerUpper)
+			if err != nil {
+				t.Errorf("routine: %v got unexpected error: %v", ver, err)
+			}
+			if diff := cmp.Diff(want, got); diff != "" {
+				t.Fatalf("routine: %v mismatch (-want, +got):\n%s", ver, diff)
+			}
+			done <- true
+		}()
+	}
+
+	for i := 0; i < 10; i++ {
+		select {
+		case <-done:
+			continue
+		case <-time.After(1 * time.Second):
+			t.Fatal("gorountines didn't termine fast enough")
+		}
+	}
+
+	if lookupCount != 1 {
+		t.Errorf("unexpected lookupCount, want: 1, got: %v", lookupCount)
+	}
+}