Implement a caching blob store

blob/cache/store.go

@@ -0,0 +1,91 @@
+package cache
+
+import (
+	"context"
+
+	"github.com/ReneKroon/ttlcache"
+
+	blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1"
+
+	"github.com/code-payments/flipcash2-server/blob"
+)
+
+// Cache wraps a blob.Store, caching only blobs that have reached a terminal
+// state (Ready or Rejected).
+type Cache struct {
+	db    blob.Store
+	blobs *ttlcache.Cache
+}
+
+// NewInCache returns a blob.Store that caches terminal blob records in front of
+// db. It composes over any backing implementation.
+func NewInCache(db blob.Store) blob.Store {
+	return &Cache{
+		db:    db,
+		blobs: ttlcache.NewCache(),
+	}
+}
+
+func (c *Cache) CreatePending(ctx context.Context, b *blob.Blob) error {
+	return c.db.CreatePending(ctx, b)
+}
+
+func (c *Cache) GetByID(ctx context.Context, id *blobpb.BlobId) (*blob.Blob, error) {
+	key := string(id.Value)
+	if cached, ok := c.blobs.Get(key); ok {
+		return cached.(*blob.Blob).Clone(), nil
+	}
+
+	b, err := c.db.GetByID(ctx, id)
+	if err == nil && b.State.Terminal() {
+		c.blobs.Set(key, b.Clone())
+	}
+	return b, err
+}
+
+func (c *Cache) GetByIDs(ctx context.Context, ids []*blobpb.BlobId) ([]*blob.Blob, error) {
+	out := make([]*blob.Blob, 0, len(ids))
+	seen := make(map[string]struct{}, len(ids))
+	misses := make([]*blobpb.BlobId, 0, len(ids))
+	for _, id := range ids {
+		key := string(id.Value)
+		// A BlobId resolves to at most one record, so collapse duplicate ids to a
+		// single result, matching the underlying store's contract.
+		if _, dup := seen[key]; dup {
+			continue
+		}
+		seen[key] = struct{}{}
+
+		if cached, ok := c.blobs.Get(key); ok {
+			out = append(out, cached.(*blob.Blob).Clone())
+			continue
+		}
+		misses = append(misses, id)
+	}
+
+	if len(misses) > 0 {
+		fetched, err := c.db.GetByIDs(ctx, misses)
+		if err != nil {
+			return nil, err
+		}
+		for _, b := range fetched {
+			if b.State.Terminal() {
+				c.blobs.Set(string(b.ID.Value), b.Clone())
+			}
+			out = append(out, b)
+		}
+	}
+	return out, nil
+}
+
+func (c *Cache) GetRenditions(ctx context.Context, parentID *blobpb.BlobId) ([]*blob.Blob, error) {
+	return c.db.GetRenditions(ctx, parentID)
+}
+
+func (c *Cache) Advance(ctx context.Context, id *blobpb.BlobId, to blob.State, image *blob.ImageMetadata) (bool, error) {
+	return c.db.Advance(ctx, id, to, image)
+}
+
+func (c *Cache) Reject(ctx context.Context, id *blobpb.BlobId, rejection *blob.RejectionMetadata) (bool, error) {
+	return c.db.Reject(ctx, id, rejection)
+}

blob/cache/store_test.go

@@ -0,0 +1,220 @@
+package cache_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1"
+
+	"github.com/code-payments/flipcash2-server/blob"
+	"github.com/code-payments/flipcash2-server/blob/cache"
+	"github.com/code-payments/flipcash2-server/blob/memory"
+	"github.com/code-payments/flipcash2-server/blob/tests"
+	"github.com/code-payments/flipcash2-server/model"
+)
+
+func TestBlob_CachingStore(t *testing.T) {
+	tests.RunStoreTests(t, cache.NewInCache(memory.NewInMemory()), func() {})
+}
+
+// countingStore is a blob.Store that counts read calls (and how many ids each
+// GetByIDs actually fetched) while delegating to a real backing store, so the
+// tests can prove which reads were served from cache versus the backing.
+type countingStore struct {
+	blob.Store
+
+	mu         sync.Mutex
+	getByID    int
+	getByIDs   int
+	idsFetched int
+}
+
+func (s *countingStore) GetByID(ctx context.Context, id *blobpb.BlobId) (*blob.Blob, error) {
+	s.mu.Lock()
+	s.getByID++
+	s.mu.Unlock()
+	return s.Store.GetByID(ctx, id)
+}
+
+func (s *countingStore) GetByIDs(ctx context.Context, ids []*blobpb.BlobId) ([]*blob.Blob, error) {
+	s.mu.Lock()
+	s.getByIDs++
+	s.idsFetched += len(ids)
+	s.mu.Unlock()
+	return s.Store.GetByIDs(ctx, ids)
+}
+
+func (s *countingStore) counts() (getByID, getByIDs, idsFetched int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.getByID, s.getByIDs, s.idsFetched
+}
+
+func newCache() (*countingStore, blob.Store) {
+	backing := &countingStore{Store: memory.NewInMemory()}
+	return backing, cache.NewInCache(backing)
+}
+
+func newPending(t *testing.T) *blob.Blob {
+	id := blob.MustGenerateID()
+	key, err := blob.StorageKey(id, "image/png")
+	require.NoError(t, err)
+	return &blob.Blob{
+		ID:         id,
+		Rendition:  blob.RenditionOriginal,
+		Owner:      model.MustGenerateUserID(),
+		State:      blob.StatePending,
+		StorageKey: key,
+		MimeType:   "image/png",
+		SizeBytes:  1234,
+	}
+}
+
+// seedReady creates a blob and advances it to the terminal READY state through
+// the given store.
+func seedReady(t *testing.T, ctx context.Context, store blob.Store) *blob.Blob {
+	b := newPending(t)
+	require.NoError(t, store.CreatePending(ctx, b))
+	for _, to := range []blob.State{blob.StateUploaded, blob.StateInspected, blob.StatePromoted, blob.StateReady} {
+		_, err := store.Advance(ctx, b.ID, to, nil)
+		require.NoError(t, err)
+	}
+	return b
+}
+
+func TestCache_GetByID_CachesTerminal(t *testing.T) {
+	ctx := context.Background()
+	backing, c := newCache()
+
+	b := seedReady(t, ctx, c)
+
+	for range 3 {
+		got, err := c.GetByID(ctx, b.ID)
+		require.NoError(t, err)
+		require.Equal(t, blob.StateReady, got.State)
+	}
+
+	// A terminal blob is immutable, so it is fetched once and served from cache
+	// thereafter.
+	getByID, _, _ := backing.counts()
+	require.Equal(t, 1, getByID)
+}
+
+func TestCache_GetByID_DoesNotCacheNonTerminalAndReflectsAdvance(t *testing.T) {
+	ctx := context.Background()
+	backing, c := newCache()
+
+	b := newPending(t)
+	require.NoError(t, c.CreatePending(ctx, b))
+
+	// A non-terminal (pending) blob is re-queried every time, never cached.
+	for range 2 {
+		got, err := c.GetByID(ctx, b.ID)
+		require.NoError(t, err)
+		require.Equal(t, blob.StatePending, got.State)
+	}
+	getByID, _, _ := backing.counts()
+	require.Equal(t, 2, getByID)
+
+	// Once it advances to a terminal state, the read reflects the new state (no
+	// stale pending is served) and is cached from then on.
+	for _, to := range []blob.State{blob.StateUploaded, blob.StateInspected, blob.StatePromoted, blob.StateReady} {
+		_, err := c.Advance(ctx, b.ID, to, nil)
+		require.NoError(t, err)
+	}
+
+	got, err := c.GetByID(ctx, b.ID)
+	require.NoError(t, err)
+	require.Equal(t, blob.StateReady, got.State) // fetched fresh, now cached
+
+	got, err = c.GetByID(ctx, b.ID)
+	require.NoError(t, err)
+	require.Equal(t, blob.StateReady, got.State) // served from cache
+
+	getByID, _, _ = backing.counts()
+	require.Equal(t, 3, getByID)
+}
+
+func TestCache_GetByIDs_ServesHitsAndFetchesOnlyMisses(t *testing.T) {
+	ctx := context.Background()
+	backing, c := newCache()
+
+	ready1 := seedReady(t, ctx, c)
+	ready2 := seedReady(t, ctx, c)
+	pending := newPending(t)
+	require.NoError(t, c.CreatePending(ctx, pending))
+	missing := blob.MustGenerateID()
+
+	// Warm the cache for ready1.
+	_, err := c.GetByID(ctx, ready1.ID)
+	require.NoError(t, err)
+
+	// A mixed batch: a cache hit (ready1), an uncached terminal (ready2), an
+	// uncached non-terminal (pending), and a missing id.
+	got, err := c.GetByIDs(ctx, []*blobpb.BlobId{ready1.ID, ready2.ID, pending.ID, missing})
+	require.NoError(t, err)
+	require.Len(t, got, 3) // the missing id is omitted
+
+	// ready1 was served from cache, so only the three misses reach the backing.
+	_, getByIDs, idsFetched := backing.counts()
+	require.Equal(t, 1, getByIDs)
+	require.Equal(t, 3, idsFetched)
+
+	// ready2 is now cached (terminal); pending and missing are not. A second batch
+	// of the two terminals is served entirely from cache.
+	got, err = c.GetByIDs(ctx, []*blobpb.BlobId{ready1.ID, ready2.ID})
+	require.NoError(t, err)
+	require.Len(t, got, 2)
+	_, getByIDs, _ = backing.counts()
+	require.Equal(t, 1, getByIDs) // unchanged: no backing call
+
+	// The still-pending blob keeps hitting the backing.
+	_, err = c.GetByIDs(ctx, []*blobpb.BlobId{pending.ID})
+	require.NoError(t, err)
+	_, getByIDs, _ = backing.counts()
+	require.Equal(t, 2, getByIDs)
+}
+
+func TestCache_GetByIDs_CollapsesDuplicates(t *testing.T) {
+	ctx := context.Background()
+	_, c := newCache()
+
+	ready := seedReady(t, ctx, c)
+
+	// A batch repeating the same id resolves to a single record, whether served
+	// from the backing (first call) or the cache (second call).
+	got, err := c.GetByIDs(ctx, []*blobpb.BlobId{ready.ID, ready.ID})
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+
+	got, err = c.GetByIDs(ctx, []*blobpb.BlobId{ready.ID, ready.ID})
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+}
+
+func TestCache_GetByID_IsolatesCachedValueFromMutation(t *testing.T) {
+	ctx := context.Background()
+	backing, c := newCache()
+
+	b := seedReady(t, ctx, c)
+
+	// First read populates the cache; mutate the returned copy in place.
+	got, err := c.GetByID(ctx, b.ID)
+	require.NoError(t, err)
+	got.MimeType = "mutated"
+	got.Owner.Value[0] ^= 0xff
+
+	// A subsequent read returns the pristine cached record, unaffected by the
+	// caller's mutation.
+	again, err := c.GetByID(ctx, b.ID)
+	require.NoError(t, err)
+	require.Equal(t, "image/png", again.MimeType)
+	require.Equal(t, b.Owner.Value, again.Owner.Value)
+
+	// And it was served from cache (a single backing fetch).
+	getByID, _, _ := backing.counts()
+	require.Equal(t, 1, getByID)
+}