add cache flush API

Author: Thomas Stromberg

cache.go

@@ -235,6 +235,25 @@ func (c *Cache[K, V]) Cleanup() int {
 	return c.memory.cleanupMemory()
 }
 
+// Flush removes all entries from the cache, including persistent storage.
+// Returns the total number of entries removed from memory and persistence.
+func (c *Cache[K, V]) Flush(ctx context.Context) (int, error) {
+	memoryRemoved := c.memory.flushMemory()
+
+	if c.persist == nil {
+		slog.Info("cache flushed", "memory", memoryRemoved, "persist", 0)
+		return memoryRemoved, nil
+	}
+
+	persistRemoved, err := c.persist.Flush(ctx)
+	if err != nil {
+		return memoryRemoved, fmt.Errorf("persistence flush failed: %w", err)
+	}
+
+	slog.Info("cache flushed", "memory", memoryRemoved, "persist", persistRemoved)
+	return memoryRemoved + persistRemoved, nil
+}
+
 // Len returns the number of items in the memory cache.
 func (c *Cache[K, V]) Len() int {
 	return c.memory.memoryLen()

cache_persist_test.go

@@ -149,6 +149,15 @@ func (m *mockPersister[K, V]) Location(key K) string {
 	return fmt.Sprintf("mock://%v", key)
 }
 
+func (m *mockPersister[K, V]) Flush(ctx context.Context) (int, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	count := len(m.data)
+	m.data = make(map[string]mockEntry[V])
+	return count, nil
+}
+
 func (m *mockPersister[K, V]) Close() error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -647,3 +656,66 @@ func TestCache_CleanupExpiredEntries(t *testing.T) {
 		t.Error("valid entry should still exist after cleanup")
 	}
 }
+
+func TestCache_FlushWithPersistence(t *testing.T) {
+	ctx := context.Background()
+	persister := newMockPersister[string, int]()
+
+	cache, err := New[string, int](ctx, WithPersistence(persister))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer func() { _ = cache.Close() }() //nolint:errcheck // Test cleanup
+
+	// Add entries
+	for i := range 10 {
+		if err := cache.Set(ctx, fmt.Sprintf("key%d", i), i*100, 0); err != nil {
+			t.Fatalf("Set: %v", err)
+		}
+	}
+
+	// Verify entries exist in both memory and persistence
+	if cache.Len() != 10 {
+		t.Errorf("memory cache length = %d; want 10", cache.Len())
+	}
+	for i := range 10 {
+		if _, _, found, err := persister.Load(ctx, fmt.Sprintf("key%d", i)); err != nil || !found {
+			t.Fatalf("key%d should exist in persistence", i)
+		}
+	}
+
+	// Flush
+	removed, err := cache.Flush(ctx)
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	// Should remove 10 from memory + 10 from persistence = 20 total
+	if removed != 20 {
+		t.Errorf("Flush removed %d items; want 20", removed)
+	}
+
+	// Memory cache should be empty
+	if cache.Len() != 0 {
+		t.Errorf("memory cache length after flush = %d; want 0", cache.Len())
+	}
+
+	// Persistence should be empty
+	for i := range 10 {
+		if _, _, found, err := persister.Load(ctx, fmt.Sprintf("key%d", i)); err != nil {
+			t.Fatalf("Load: %v", err)
+		} else if found {
+			t.Errorf("key%d should not exist in persistence after flush", i)
+		}
+	}
+
+	// Get should return not found for all keys
+	for i := range 10 {
+		_, found, err := cache.Get(ctx, fmt.Sprintf("key%d", i))
+		if err != nil {
+			t.Fatalf("Get: %v", err)
+		}
+		if found {
+			t.Errorf("key%d should not be found after flush", i)
+		}
+	}
+}

cache_test.go

@@ -2,6 +2,7 @@ package bdcache
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -625,3 +626,66 @@ func TestCache_SetUpdateExisting(t *testing.T) {
 		t.Errorf("Get value = %d; want 100", val)
 	}
 }
+
+func TestCache_Flush(t *testing.T) {
+	ctx := context.Background()
+	cache, err := New[string, int](ctx)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer func() { _ = cache.Close() }() //nolint:errcheck // Test cleanup
+
+	// Add entries
+	for i := range 10 {
+		if err := cache.Set(ctx, fmt.Sprintf("key%d", i), i, 0); err != nil {
+			t.Fatalf("Set: %v", err)
+		}
+	}
+
+	if cache.Len() != 10 {
+		t.Errorf("cache length = %d; want 10", cache.Len())
+	}
+
+	// Flush
+	removed, err := cache.Flush(ctx)
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if removed != 10 {
+		t.Errorf("Flush removed %d items; want 10", removed)
+	}
+
+	// Cache should be empty
+	if cache.Len() != 0 {
+		t.Errorf("cache length after flush = %d; want 0", cache.Len())
+	}
+
+	// All keys should be gone
+	for i := range 10 {
+		_, found, err := cache.Get(ctx, fmt.Sprintf("key%d", i))
+		if err != nil {
+			t.Fatalf("Get: %v", err)
+		}
+		if found {
+			t.Errorf("key%d should not be found after flush", i)
+		}
+	}
+}
+
+func TestCache_FlushEmpty(t *testing.T) {
+	ctx := context.Background()
+	cache, err := New[string, int](ctx)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer func() { _ = cache.Close() }() //nolint:errcheck // Test cleanup
+
+	// Flush empty cache
+	removed, err := cache.Flush(ctx)
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if removed != 0 {
+		t.Errorf("Flush removed %d items; want 0", removed)
+	}
+}

persist.go

@@ -39,6 +39,10 @@ type PersistenceLayer[K comparable, V any] interface {
 	// Useful for testing and debugging to verify where items are stored.
 	Location(key K) string
 
+	// Flush removes all entries from persistent storage.
+	// Returns the number of entries removed and any error.
+	Flush(ctx context.Context) (int, error)
+
 	// Close releases any resources held by the persistence layer.
 	Close() error
 }

persist/datastore/datastore.go

@@ -281,6 +281,31 @@ func (p *persister[K, V]) Cleanup(ctx context.Context, maxAge time.Duration) (in
 	return len(keys), nil
 }
 
+// Flush removes all entries from Datastore.
+// Returns the number of entries removed and any error.
+func (p *persister[K, V]) Flush(ctx context.Context) (int, error) {
+	// Query for all keys
+	query := ds.NewQuery(p.kind).KeysOnly()
+
+	var entries []datastoreEntry
+	keys, err := p.client.GetAll(ctx, query, &entries)
+	if err != nil {
+		return 0, fmt.Errorf("query all entries: %w", err)
+	}
+
+	if len(keys) == 0 {
+		return 0, nil
+	}
+
+	// Batch delete all entries
+	if err := p.client.DeleteMulti(ctx, keys); err != nil {
+		return 0, fmt.Errorf("delete all entries: %w", err)
+	}
+
+	slog.Info("flushed datastore cache", "count", len(keys), "kind", p.kind)
+	return len(keys), nil
+}
+
 // Close releases Datastore client resources.
 func (p *persister[K, V]) Close() error {
 	return p.client.Close()

persist/datastore/persist_datastore_mock_test.go

@@ -3,6 +3,7 @@ package datastore
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -770,3 +771,61 @@ func TestDatastorePersist_Mock_MultipleOps(t *testing.T) {
 		t.Error("After Delete: key should not be found")
 	}
 }
+
+func TestDatastorePersist_Mock_Flush(t *testing.T) {
+	dp, cleanup := newMockDatastorePersist[string, int](t)
+	defer cleanup()
+
+	ctx := context.Background()
+
+	// Store multiple entries
+	for i := range 10 {
+		key := fmt.Sprintf("key%d", i)
+		if err := dp.Store(ctx, key, i*100, time.Time{}); err != nil {
+			t.Fatalf("Store %s: %v", key, err)
+		}
+	}
+
+	// Verify entries exist
+	for i := range 10 {
+		key := fmt.Sprintf("key%d", i)
+		if _, _, found, err := dp.Load(ctx, key); err != nil || !found {
+			t.Fatalf("%s should exist before flush", key)
+		}
+	}
+
+	// Flush
+	deleted, err := dp.Flush(ctx)
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if deleted != 10 {
+		t.Errorf("Flush deleted %d entries; want 10", deleted)
+	}
+
+	// All entries should be gone
+	for i := range 10 {
+		key := fmt.Sprintf("key%d", i)
+		if _, _, found, err := dp.Load(ctx, key); err != nil {
+			t.Fatalf("Load: %v", err)
+		} else if found {
+			t.Errorf("%s should not exist after flush", key)
+		}
+	}
+}
+
+func TestDatastorePersist_Mock_FlushEmpty(t *testing.T) {
+	dp, cleanup := newMockDatastorePersist[string, int](t)
+	defer cleanup()
+
+	ctx := context.Background()
+
+	// Flush empty datastore
+	deleted, err := dp.Flush(ctx)
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if deleted != 0 {
+		t.Errorf("Flush deleted %d entries from empty datastore; want 0", deleted)
+	}
+}

persist/localfs/integration_test.go

@@ -902,3 +902,102 @@ func TestFilePersist_Cleanup_ContextCancellation(t *testing.T) {
 		t.Errorf("expected context.Canceled error, got: %v", err)
 	}
 }
+
+func TestFilePersist_Flush(t *testing.T) {
+	dir := t.TempDir()
+	fp, err := New[string, int]("test", dir)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer func() {
+		if err := fp.Close(); err != nil {
+			t.Logf("Close error: %v", err)
+		}
+	}()
+
+	ctx := context.Background()
+
+	// Store multiple entries
+	for i := range 10 {
+		if err := fp.Store(ctx, fmt.Sprintf("key-%d", i), i*100, time.Time{}); err != nil {
+			t.Fatalf("Store: %v", err)
+		}
+	}
+
+	// Verify files exist
+	for i := range 10 {
+		if _, _, found, err := fp.Load(ctx, fmt.Sprintf("key-%d", i)); err != nil || !found {
+			t.Fatalf("key-%d should exist before flush", i)
+		}
+	}
+
+	// Flush
+	deleted, err := fp.Flush(ctx)
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if deleted != 10 {
+		t.Errorf("Flush deleted %d entries; want 10", deleted)
+	}
+
+	// All entries should be gone
+	for i := range 10 {
+		if _, _, found, err := fp.Load(ctx, fmt.Sprintf("key-%d", i)); err != nil {
+			t.Fatalf("Load: %v", err)
+		} else if found {
+			t.Errorf("key-%d should not exist after flush", i)
+		}
+	}
+}
+
+func TestFilePersist_Flush_Empty(t *testing.T) {
+	dir := t.TempDir()
+	fp, err := New[string, int]("test", dir)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer func() {
+		if err := fp.Close(); err != nil {
+			t.Logf("Close error: %v", err)
+		}
+	}()
+
+	// Flush empty cache
+	deleted, err := fp.Flush(context.Background())
+	if err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if deleted != 0 {
+		t.Errorf("Flush deleted %d entries; want 0", deleted)
+	}
+}
+
+func TestFilePersist_Flush_ContextCancellation(t *testing.T) {
+	dir := t.TempDir()
+	fp, err := New[string, int]("test", dir)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer func() {
+		if err := fp.Close(); err != nil {
+			t.Logf("Close error: %v", err)
+		}
+	}()
+
+	// Store many entries
+	for i := range 100 {
+		if err := fp.Store(context.Background(), fmt.Sprintf("key-%d", i), i, time.Time{}); err != nil {
+			t.Fatalf("Store: %v", err)
+		}
+	}
+
+	// Create context that we'll cancel
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	// Try to flush
+	_, err = fp.Flush(ctx)
+	if err == nil || !errors.Is(err, context.Canceled) {
+		t.Errorf("expected context.Canceled error, got: %v", err)
+	}
+}