unified-storage: HappyPath and notifier tests + couple of bugfixes (#116087)

* unified-storage: couple of bugfixes and enable HappyPath and notifier sqlkv tests

pkg/storage/unified/resource/notifier.go

@@ -78,13 +78,13 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event {
 	cache := gocache.New(cacheTTL, cacheCleanupInterval)
 	events := make(chan Event, opts.BufferSize)
 
-	initialRV, err := n.lastEventResourceVersion(ctx)
+	lastRV, err := n.lastEventResourceVersion(ctx)
 	if errors.Is(err, ErrNotFound) {
-		initialRV = snowflakeFromTime(time.Now()) // No events yet, start from the beginning
+		lastRV = 0 // No events yet, start from the beginning
 	} else if err != nil {
 		n.log.Error("Failed to get last event resource version", "error", err)
 	}
-	lastRV := initialRV + 1 // We want to start watching from the next event
+	lastRV = lastRV + 1 // We want to start watching from the next event
 
 	go func() {
 		defer close(events)
@@ -110,7 +110,7 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event {
 					}
 
 					// Skip old events lower than the requested resource version
-					if evt.ResourceVersion <= initialRV {
+					if evt.ResourceVersion < lastRV {
 						continue
 					}
 

pkg/storage/unified/resource/notifier_test.go

@@ -25,7 +25,6 @@ func setupTestNotifier(t *testing.T) (*notifier, *eventStore) {
 	return notifier, eventStore
 }
 
-// nolint:unused
 func setupTestNotifierSqlKv(t *testing.T) (*notifier, *eventStore) {
 	dbstore := db.InitTestDB(t)
 	eDB, err := dbimpl.ProvideResourceDB(dbstore, setting.NewCfg(), nil)
@@ -60,8 +59,7 @@ func runNotifierTestWith(t *testing.T, storeName string, newStoreFn func(*testin
 
 func TestNotifier_lastEventResourceVersion(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierLastEventResourceVersion)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierLastEventResourceVersion)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierLastEventResourceVersion)
 }
 
 func testNotifierLastEventResourceVersion(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {
@@ -112,8 +110,7 @@ func testNotifierLastEventResourceVersion(t *testing.T, ctx context.Context, not
 
 func TestNotifier_cachekey(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierCachekey)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierCachekey)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierCachekey)
 }
 
 func testNotifierCachekey(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {
@@ -167,8 +164,7 @@ func testNotifierCachekey(t *testing.T, ctx context.Context, notifier *notifier,
 
 func TestNotifier_Watch_NoEvents(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchNoEvents)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchNoEvents)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchNoEvents)
 }
 
 func testNotifierWatchNoEvents(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {
@@ -209,8 +205,7 @@ func testNotifierWatchNoEvents(t *testing.T, ctx context.Context, notifier *noti
 
 func TestNotifier_Watch_WithExistingEvents(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchWithExistingEvents)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchWithExistingEvents)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchWithExistingEvents)
 }
 
 func testNotifierWatchWithExistingEvents(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {
@@ -284,8 +279,7 @@ func testNotifierWatchWithExistingEvents(t *testing.T, ctx context.Context, noti
 
 func TestNotifier_Watch_EventDeduplication(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchEventDeduplication)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchEventDeduplication)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchEventDeduplication)
 }
 
 func testNotifierWatchEventDeduplication(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {
@@ -351,8 +345,7 @@ func testNotifierWatchEventDeduplication(t *testing.T, ctx context.Context, noti
 
 func TestNotifier_Watch_ContextCancellation(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchContextCancellation)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchContextCancellation)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchContextCancellation)
 }
 
 func testNotifierWatchContextCancellation(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {
@@ -398,8 +391,7 @@ func testNotifierWatchContextCancellation(t *testing.T, ctx context.Context, not
 
 func TestNotifier_Watch_MultipleEvents(t *testing.T) {
 	runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchMultipleEvents)
-	// enable this when sqlkv is ready
-	// runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchMultipleEvents)
+	runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchMultipleEvents)
 }
 
 func testNotifierWatchMultipleEvents(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) {

pkg/storage/unified/resource/storage_backend.go

@@ -346,7 +346,8 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in
 			return 0, fmt.Errorf("failed to write data: %w", err)
 		}
 
-		dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(rv)
+		rv = rvmanager.SnowflakeFromRv(rv)
+		dataKey.ResourceVersion = rv
 	} else {
 		err := k.dataStore.Save(ctx, dataKey, bytes.NewReader(event.Value))
 		if err != nil {

pkg/storage/unified/testing/storage_backend_sql_compatibility.go

@@ -9,7 +9,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/bwmarrin/snowflake"
 	"github.com/stretchr/testify/require"
 
 	claims "github.com/grafana/authlib/types"
@@ -187,13 +186,30 @@ func runKeyPathTest(t *testing.T, backend resource.StorageBackend, nsPrefix stri
 
 // verifyKeyPath is a helper function to verify key_path generation
 func verifyKeyPath(t *testing.T, db sqldb.DB, ctx context.Context, key *resourcepb.ResourceKey, action string, resourceVersion int64, expectedFolder string) {
+	// For SQL backend (namespace contains "-sql"), resourceVersion is in microsecond format
+	// but key_path stores snowflake RV, so convert to snowflake
+	// For KV backend (namespace contains "-kv"), resourceVersion is already in snowflake format
+	isSqlBackend := strings.Contains(key.Namespace, "-sql")
+
+	var keyPathRV int64
+	if isSqlBackend {
+		// Convert microsecond RV to snowflake for key_path construction
+		keyPathRV = rvmanager.SnowflakeFromRv(resourceVersion)
+	} else {
+		// KV backend already provides snowflake RV
+		keyPathRV = resourceVersion
+	}
+
+	// Build the expected key_path using DataKey format: unified/data/group/resource/namespace/name/resourceVersion~action~folder
+	expectedKeyPath := fmt.Sprintf("unified/data/%s/%s/%s/%s/%d~%s~%s", key.Group, key.Resource, key.Namespace, key.Name, keyPathRV, action, expectedFolder)
+
 	var query string
 	if db.DriverName() == "postgres" {
-		query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE namespace = $1 AND name = $2 AND resource_version = $3"
+		query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE key_path = $1"
 	} else {
-		query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE namespace = ? AND name = ? AND resource_version = ?"
+		query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE key_path = ?"
 	}
-	rows, err := db.QueryContext(ctx, query, key.Namespace, key.Name, resourceVersion)
+	rows, err := db.QueryContext(ctx, query, expectedKeyPath)
 	require.NoError(t, err)
 
 	require.True(t, rows.Next(), "Resource not found in resource_history table - both SQL and KV backends should write to this table")
@@ -220,10 +236,6 @@ func verifyKeyPath(t *testing.T, db sqldb.DB, ctx context.Context, key *resource
 	// Verify action suffix
 	require.Contains(t, keyPath, fmt.Sprintf("~%s~", action))
 
-	// Verify snowflake calculation
-	expectedSnowflake := (((resourceVersion / 1000) - snowflake.Epoch) << (snowflake.NodeBits + snowflake.StepBits)) + (resourceVersion % 1000)
-	require.Contains(t, keyPath, fmt.Sprintf("/%d~", expectedSnowflake), "actual RV: %d", actualRV)
-
 	// Verify folder if specified
 	if expectedFolder != "" {
 		require.Equal(t, expectedFolder, actualFolder)
@@ -492,10 +504,10 @@ func verifyResourceHistoryRecord(t *testing.T, record ResourceHistoryRecord, exp
 	}
 
 	// Validate previous_resource_version
-	// For KV backend operations, resource versions are stored as snowflake format
-	// but expectedPrevRV is in microsecond format, so we need to use IsRvEqual for comparison
+	// For KV backend operations, expectedPrevRV is now in snowflake format (returned by KV backend)
+	// but resource_history table stores microsecond RV, so we need to use IsRvEqual for comparison
 	if strings.Contains(record.Namespace, "-kv") {
-		require.True(t, rvmanager.IsRvEqual(record.PreviousResourceVersion, expectedPrevRV),
+		require.True(t, rvmanager.IsRvEqual(expectedPrevRV, record.PreviousResourceVersion),
 			"Previous resource version should match (KV backend snowflake format)")
 	} else {
 		require.Equal(t, expectedPrevRV, record.PreviousResourceVersion)
@@ -505,9 +517,10 @@ func verifyResourceHistoryRecord(t *testing.T, record ResourceHistoryRecord, exp
 	require.Equal(t, expectedGeneration, record.Generation)
 
 	// Validate resource_version
-	// For KV backend operations, resource versions are stored as snowflake format
+	// For KV backend operations, expectedRV is now in snowflake format (returned by KV backend)
+	// but resource_history table stores microsecond RV, so we need to use IsRvEqual for comparison
 	if strings.Contains(record.Namespace, "-kv") {
-		require.True(t, rvmanager.IsRvEqual(record.ResourceVersion, expectedRV),
+		require.True(t, rvmanager.IsRvEqual(expectedRV, record.ResourceVersion),
 			"Resource version should match (KV backend snowflake format)")
 	} else {
 		require.Equal(t, expectedRV, record.ResourceVersion)
@@ -574,7 +587,7 @@ func verifyResourceTable(t *testing.T, db sqldb.DB, namespace string, resources
 	// Resource version should match the expected version for test-resource-3 (updated version)
 	expectedRV := resourceVersions[2][1] // test-resource-3's update version
 	if strings.Contains(namespace, "-kv") {
-		require.True(t, rvmanager.IsRvEqual(record.ResourceVersion, expectedRV),
+		require.True(t, rvmanager.IsRvEqual(expectedRV, record.ResourceVersion),
 			"Resource version should match (KV backend snowflake format)")
 	} else {
 		require.Equal(t, expectedRV, record.ResourceVersion)
@@ -625,9 +638,16 @@ func verifyResourceVersionTable(t *testing.T, db sqldb.DB, namespace string, res
 
 	// The resource_version table should contain the latest RV for the group+resource
 	// It might be slightly higher due to RV manager operations, so check it's at least our max
-	require.GreaterOrEqual(t, record.ResourceVersion, maxRV, "resource_version should be at least the latest RV we tracked")
-	// But it shouldn't be too much higher (within a reasonable range)
-	require.LessOrEqual(t, record.ResourceVersion, maxRV+100, "resource_version shouldn't be much higher than expected")
+	// For KV backend, maxRV is in snowflake format but record.ResourceVersion is in microsecond format
+	// Use IsRvEqual for proper comparison between different RV formats
+	isKvBackend := strings.Contains(namespace, "-kv")
+	recordResourceVersion := record.ResourceVersion
+	if isKvBackend {
+		recordResourceVersion = rvmanager.SnowflakeFromRv(record.ResourceVersion)
+	}
+
+	require.Less(t, recordResourceVersion, int64(9223372036854775807), "resource_version should be reasonable")
+	require.Greater(t, recordResourceVersion, maxRV, "resource_version should be at least the latest RV we tracked")
 }
 
 // runTestCrossBackendConsistency tests basic consistency between SQL and KV backends (lightweight)

pkg/storage/unified/testing/storage_backend_test.go

@@ -38,7 +38,6 @@ func TestBadgerKVStorageBackend(t *testing.T) {
 
 func TestSQLKVStorageBackend(t *testing.T) {
 	skipTests := map[string]bool{
-		TestHappyPath:                 true,
 		TestWatchWriteEvents:          true,
 		TestList:                      true,
 		TestBlobSupport:               true,
@@ -51,21 +50,24 @@ func TestSQLKVStorageBackend(t *testing.T) {
 		TestGetResourceLastImportTime: true,
 		TestOptimisticLocking:         true,
 	}
-	// without RvManager
-	RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend {
-		backend, _ := NewTestSqlKvBackend(t, ctx, false)
-		return backend
-	}, &TestOptions{
-		NSPrefix:  "sqlkvstorage-test",
-		SkipTests: skipTests,
+
+	t.Run("Without RvManager", func(t *testing.T) {
+		RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend {
+			backend, _ := NewTestSqlKvBackend(t, ctx, false)
+			return backend
+		}, &TestOptions{
+			NSPrefix:  "sqlkvstorage-test",
+			SkipTests: skipTests,
+		})
 	})
 
-	// with RvManager
-	RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend {
-		backend, _ := NewTestSqlKvBackend(t, ctx, true)
-		return backend
-	}, &TestOptions{
-		NSPrefix:  "sqlkvstorage-withrvmanager-test",
-		SkipTests: skipTests,
+	t.Run("With RvManager", func(t *testing.T) {
+		RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend {
+			backend, _ := NewTestSqlKvBackend(t, ctx, true)
+			return backend
+		}, &TestOptions{
+			NSPrefix:  "sqlkvstorage-withrvmanager-test",
+			SkipTests: skipTests,
+		})
 	})
 }