Feat: Prevent multiple identical inflight channel search requests (#16)

* add new cache type with loader to avoid duplicate concurrent requests to the Sift API

Author: alexluck-sift

pkg/plugin/cache_utils.go

@@ -2,10 +2,12 @@ package plugin
 
 import (
 	"fmt"
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/backend/log"
+	"github.com/patrickmn/go-cache"
 	"math/rand"
+	"sync"
 	"time"
-
-	"github.com/patrickmn/go-cache"
 )
 
 // TypedCache is a type-safe wrapper around go-cache
@@ -51,7 +53,6 @@ func (tc *TypedCache[K, V]) Get(key K) (V, bool) {
 	if !ok {
 		return result, false
 	}
-
 	return typedValue, true
 }
 
@@ -72,3 +73,61 @@ func (tc *TypedCache[K, V]) getRandomizedTimeToLive() time.Duration {
 	}
 	return time.Duration(rand.Intn(int(tc.maxTtl-tc.minTtl))) + tc.minTtl
 }
+
+type TypedCacheWithLoader[K any, V any, C comparable] struct {
+	*TypedCache[C, V]
+	mu              *sync.Mutex
+	keyToComparable func(K) C
+	loading         map[C]func() (V, error)
+	loader          func(*SiftDatasource, backend.PluginContext, K) (V, error)
+}
+
+func NewTypedCacheWithLoader[K any, V any, C comparable](typedCache *TypedCache[C, V], loader func(*SiftDatasource, backend.PluginContext, K) (V, error), keyToComparable func(K) C) *TypedCacheWithLoader[K, V, C] {
+	return &TypedCacheWithLoader[K, V, C]{
+		TypedCache:      typedCache,
+		mu:              &sync.Mutex{},
+		keyToComparable: keyToComparable,
+		loading:         make(map[C]func() (V, error)),
+		loader:          loader,
+	}
+}
+
+// GetOrWait retrieves an item from the cache and waits on any other goroutine that is setting the value
+func (tc *TypedCacheWithLoader[K, V, C]) GetOrWait(d *SiftDatasource, ctx backend.PluginContext, key K) (V, error) {
+	tc.mu.Lock()
+	comparableKey := tc.keyToComparable(key)
+	value, found := tc.cache.Get(fmt.Sprintf("%v", comparableKey))
+	if found {
+		if typedValue, ok := value.(V); ok {
+			tc.mu.Unlock()
+			return typedValue, nil
+		}
+	}
+
+	// Check if we are already loading the value
+	load := tc.loading[comparableKey]
+	if load != nil {
+		tc.mu.Unlock()
+		log.DefaultLogger.Debug("waiting for pending cache load", "key", comparableKey)
+		return load()
+	}
+
+	// Haven't started loading it
+	log.DefaultLogger.Debug("initiating new cache load", "key", comparableKey)
+	load = sync.OnceValues(func() (V, error) {
+		v, err := tc.loader(d, ctx, key)
+		tc.mu.Lock()
+		defer tc.mu.Unlock()
+
+		delete(tc.loading, comparableKey)
+		if err != nil {
+			return v, err
+		}
+		tc.Set(comparableKey, v)
+		return v, nil
+	})
+	tc.loading[comparableKey] = load
+	tc.mu.Unlock()
+	return load()
+
+}

pkg/plugin/cache_utils_test.go

@@ -1,8 +1,12 @@
 package plugin
 
 import (
+	"errors"
+	"sync"
 	"testing"
 	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
 )
 
 func TestTypedCache_BasicOperations(t *testing.T) {
@@ -65,3 +69,237 @@ func TestTypedCache_WithStructKeys(t *testing.T) {
 		t.Errorf("Expected 200, got %d", value)
 	}
 }
+
+func TestTypedCacheWithLoader_GetOrWait_CacheHit(t *testing.T) {
+	// Create a typed cache
+	cache := NewTypedCache[string, string](5*time.Minute, 10*time.Minute)
+	
+	// Pre-populate the cache
+	cache.Set("test-key", "cached-value")
+	
+	// Create a loader that should not be called
+	loader := func(d *SiftDatasource, ctx backend.PluginContext, key string) (string, error) {
+		t.Error("Loader should not be called when value is in cache")
+		return "", errors.New("should not be called")
+	}
+	
+	// Create cache with loader
+	cacheWithLoader := NewTypedCacheWithLoader(cache, loader, func(k string) string { return k })
+	
+	// Mock datasource and context
+	mockDatasource := &SiftDatasource{}
+	mockContext := backend.PluginContext{}
+	
+	// Test GetOrWait - should return cached value
+	result, err := cacheWithLoader.GetOrWait(mockDatasource, mockContext, "test-key")
+	
+	if err != nil {
+		t.Errorf("Expected no error, got %v", err)
+	}
+	if result != "cached-value" {
+		t.Errorf("Expected 'cached-value', got %s", result)
+	}
+}
+
+func TestTypedCacheWithLoader_GetOrWait_CacheMiss(t *testing.T) {
+	// Create an empty typed cache
+	cache := NewTypedCache[string, string](5*time.Minute, 10*time.Minute)
+	
+	// Create a loader that returns a value
+	loaderCalled := false
+	loader := func(d *SiftDatasource, ctx backend.PluginContext, key string) (string, error) {
+		loaderCalled = true
+		if key == "test-key" {
+			return "loaded-value", nil
+		}
+		return "", errors.New("unexpected key")
+	}
+	
+	// Create cache with loader
+	cacheWithLoader := NewTypedCacheWithLoader(cache, loader, func(k string) string { return k })
+	
+	// Mock datasource and context
+	mockDatasource := &SiftDatasource{}
+	mockContext := backend.PluginContext{}
+	
+	// Test GetOrWait - should call loader and return loaded value
+	result, err := cacheWithLoader.GetOrWait(mockDatasource, mockContext, "test-key")
+	
+	if err != nil {
+		t.Errorf("Expected no error, got %v", err)
+	}
+	if result != "loaded-value" {
+		t.Errorf("Expected 'loaded-value', got %s", result)
+	}
+	if !loaderCalled {
+		t.Error("Expected loader to be called")
+	}
+	
+	// Verify value was cached
+	cachedValue, found := cache.Get("test-key")
+	if !found {
+		t.Error("Expected value to be cached")
+	}
+	if cachedValue != "loaded-value" {
+		t.Errorf("Expected cached value 'loaded-value', got %s", cachedValue)
+	}
+}
+
+func TestTypedCacheWithLoader_GetOrWait_LoaderError(t *testing.T) {
+	// Create an empty typed cache
+	cache := NewTypedCache[string, string](5*time.Minute, 10*time.Minute)
+	
+	// Create a loader that returns an error
+	expectedError := errors.New("loader failed")
+	loader := func(d *SiftDatasource, ctx backend.PluginContext, key string) (string, error) {
+		return "", expectedError
+	}
+	
+	// Create cache with loader
+	cacheWithLoader := NewTypedCacheWithLoader(cache, loader, func(k string) string { return k })
+	
+	// Mock datasource and context
+	mockDatasource := &SiftDatasource{}
+	mockContext := backend.PluginContext{}
+	
+	// Test GetOrWait - should return error from loader
+	result, err := cacheWithLoader.GetOrWait(mockDatasource, mockContext, "test-key")
+	
+	if err != expectedError {
+		t.Errorf("Expected error %v, got %v", expectedError, err)
+	}
+	if result != "" {
+		t.Errorf("Expected empty result on error, got %s", result)
+	}
+	
+	// Verify value was not cached on error
+	_, found := cache.Get("test-key")
+	if found {
+		t.Error("Expected value not to be cached on loader error")
+	}
+}
+
+func TestTypedCacheWithLoader_GetOrWait_ConcurrentLoads(t *testing.T) {
+	// Create an empty typed cache
+	cache := NewTypedCache[string, string](5*time.Minute, 10*time.Minute)
+	
+	// Create a loader that simulates slow loading
+	loaderCallCount := 0
+	var loaderMutex sync.Mutex
+	loader := func(d *SiftDatasource, ctx backend.PluginContext, key string) (string, error) {
+		loaderMutex.Lock()
+		loaderCallCount++
+		loaderMutex.Unlock()
+		
+		// Simulate slow operation
+		time.Sleep(100 * time.Millisecond)
+		return "loaded-value", nil
+	}
+	
+	// Create cache with loader
+	cacheWithLoader := NewTypedCacheWithLoader(cache, loader, func(k string) string { return k })
+	
+	// Mock datasource and context
+	mockDatasource := &SiftDatasource{}
+	mockContext := backend.PluginContext{}
+	
+	// Launch multiple concurrent GetOrWait calls
+	const numGoroutines = 5
+	var wg sync.WaitGroup
+	results := make([]string, numGoroutines)
+	errors := make([]error, numGoroutines)
+	
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+		go func(index int) {
+			defer wg.Done()
+			result, err := cacheWithLoader.GetOrWait(mockDatasource, mockContext, "test-key")
+			results[index] = result
+			errors[index] = err
+		}(i)
+	}
+	
+	wg.Wait()
+	
+	// Verify all calls succeeded and returned the same value
+	for i := 0; i < numGoroutines; i++ {
+		if errors[i] != nil {
+			t.Errorf("Goroutine %d got error: %v", i, errors[i])
+		}
+		if results[i] != "loaded-value" {
+			t.Errorf("Goroutine %d got result %s, expected 'loaded-value'", i, results[i])
+		}
+	}
+	
+	// Verify loader was called only once despite multiple concurrent requests
+	loaderMutex.Lock()
+	if loaderCallCount != 1 {
+		t.Errorf("Expected loader to be called exactly once, but was called %d times", loaderCallCount)
+	}
+	loaderMutex.Unlock()
+	
+	// Verify value was cached
+	cachedValue, found := cache.Get("test-key")
+	if !found {
+		t.Error("Expected value to be cached")
+	}
+	if cachedValue != "loaded-value" {
+		t.Errorf("Expected cached value 'loaded-value', got %s", cachedValue)
+	}
+}
+
+func TestTypedCacheWithLoader_GetOrWait_WithComplexTypes(t *testing.T) {
+	// Test with complex key and value types
+	type TestKey struct {
+		AssetID string
+		RunID   string
+	}
+	
+	type TestValue struct {
+		Data      string
+		Timestamp time.Time
+	}
+	
+	// Create cache with complex types
+	cache := NewTypedCache[string, TestValue](5*time.Minute, 10*time.Minute)
+	
+	// Create a loader
+	loader := func(d *SiftDatasource, ctx backend.PluginContext, key TestKey) (TestValue, error) {
+		return TestValue{
+			Data:      "complex-data-" + key.AssetID,
+			Timestamp: time.Now(),
+		}, nil
+	}
+	
+	// Key conversion function
+	keyToComparable := func(k TestKey) string {
+		return k.AssetID + ":" + k.RunID
+	}
+	
+	// Create cache with loader
+	cacheWithLoader := NewTypedCacheWithLoader(cache, loader, keyToComparable)
+	
+	// Mock datasource and context
+	mockDatasource := &SiftDatasource{}
+	mockContext := backend.PluginContext{}
+	
+	// Test with complex key
+	testKey := TestKey{AssetID: "asset123", RunID: "run456"}
+	result, err := cacheWithLoader.GetOrWait(mockDatasource, mockContext, testKey)
+	
+	if err != nil {
+		t.Errorf("Expected no error, got %v", err)
+	}
+	if result.Data != "complex-data-asset123" {
+		t.Errorf("Expected 'complex-data-asset123', got %s", result.Data)
+	}
+	
+	// Verify caching with the converted key
+	cachedValue, found := cache.Get("asset123:run456")
+	if !found {
+		t.Error("Expected value to be cached with converted key")
+	}
+	if cachedValue.Data != "complex-data-asset123" {
+		t.Errorf("Expected cached data 'complex-data-asset123', got %s", cachedValue.Data)
+	}
+}