feat(selector): surgical cache updates via extended token notifier

- Extend TokenRecordReference with WalletID, Type, Quantity
- Update Postgres trigger to include owner_wallet_id, token_type, quantity in NOTIFY payload
- Replace dirty-flag invalidation with surgical insert/delete in onTokenChange
- Add 8KB payload guard to trigger (RAISE EXCEPTION on overflow)
- Add TokenNotifierDisabled config flag for operators to opt out
- Add BenchmarkSelectorWit
Signed-off-by: Atharva Borade <atharvaborade568@gmail.com>

Signed-off-by: atharrva01 <atharvaborade568@gmail.com>

Author: atharrva01

token/sdk/dig/sdk.go

@@ -167,6 +167,7 @@ func (p *SDK) Install() error {
 				cfg.GetFetcherCacheSize(),
 				cfg.GetFetcherCacheRefresh(),
 				cfg.GetFetcherCacheMaxQueries(),
+				cfg.IsTokenNotifierDisabled(),
 			)
 		}),
 

token/services/selector/config/driver.go

@@ -38,6 +38,10 @@ type Config struct {
 	FetcherCacheSize       int64         `yaml:"fetcherCacheSize,omitempty"`
 	FetcherCacheRefresh    time.Duration `yaml:"fetcherCacheRefresh,omitempty"`
 	FetcherCacheMaxQueries int           `yaml:"fetcherCacheMaxQueries,omitempty"`
+	// TokenNotifierDisabled disables the Postgres LISTEN/NOTIFY-based cache
+	// invalidation and falls back to the time-based freshness interval.
+	// Set to true if the notifier becomes a bottleneck under high write load.
+	TokenNotifierDisabled bool `yaml:"tokenNotifierDisabled,omitempty"`
 }
 
 // New returns a SelectorConfig with the values from the token.selector key
@@ -105,3 +109,9 @@ func (c *Config) GetFetcherCacheMaxQueries() int {
 	// Return 0 if not set, which will trigger use of fetcher default
 	return c.FetcherCacheMaxQueries
 }
+
+// IsTokenNotifierDisabled returns true when the DB notifier should be skipped
+// and the selector falls back to the time-based freshness interval.
+func (c *Config) IsTokenNotifierDisabled() bool {
+	return c.TokenNotifierDisabled
+}

token/services/selector/sherdlock/fetcher.go

@@ -19,7 +19,7 @@ import (
 	"github.com/hyperledger-labs/fabric-token-sdk/token/driver"
 	dbdriver "github.com/hyperledger-labs/fabric-token-sdk/token/services/storage/db/driver"
 	"github.com/hyperledger-labs/fabric-token-sdk/token/services/storage/tokendb"
-	"github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/cache"
+	utilcache "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/cache"
 	token2 "github.com/hyperledger-labs/fabric-token-sdk/token/token"
 )
 
@@ -38,7 +38,7 @@ const (
 	Cached   FetcherStrategy = "cached"
 )
 
-type fetchFunc func(db *tokendb.StoreService, m *Metrics, cacheSize int64, freshnessInterval time.Duration, maxQueries int) TokenFetcher
+type fetchFunc func(db *tokendb.StoreService, m *Metrics, cacheSize int64, freshnessInterval time.Duration, maxQueries int, notifierDisabled bool) TokenFetcher
 
 type fetcherProvider struct {
 	tokenStoreServiceManager tokendb.StoreServiceManager
@@ -47,12 +47,13 @@ type fetcherProvider struct {
 	cacheSize                int64
 	freshnessInterval        time.Duration
 	maxQueries               int
+	notifierDisabled         bool
 }
 
 var fetchers = map[FetcherStrategy]fetchFunc{
-	Mixed: func(db *tokendb.StoreService, m *Metrics, cacheSize int64, freshnessInterval time.Duration, maxQueries int) TokenFetcher {
+	Mixed: func(db *tokendb.StoreService, m *Metrics, cacheSize int64, freshnessInterval time.Duration, maxQueries int, notifierDisabled bool) TokenFetcher {
 		var notifier dbdriver.TokenNotifier
-		if db != nil && db.TokenStore != nil {
+		if !notifierDisabled && db != nil && db.TokenStore != nil {
 			var err error
 			notifier, err = db.Notifier()
 			if err != nil {
@@ -65,7 +66,7 @@ var fetchers = map[FetcherStrategy]fetchFunc{
 }
 
 // NewFetcherProvider creates a new fetcher provider with the specified strategy and configuration.
-func NewFetcherProvider(storeServiceManager tokendb.StoreServiceManager, metricsProvider metrics.Provider, strategy FetcherStrategy, cacheSize int64, freshnessInterval time.Duration, maxQueries int) *fetcherProvider {
+func NewFetcherProvider(storeServiceManager tokendb.StoreServiceManager, metricsProvider metrics.Provider, strategy FetcherStrategy, cacheSize int64, freshnessInterval time.Duration, maxQueries int, notifierDisabled bool) *fetcherProvider {
 	fetcher, ok := fetchers[strategy]
 	if !ok {
 		panic("undefined fetcher strategy: " + strategy)
@@ -78,6 +79,7 @@ func NewFetcherProvider(storeServiceManager tokendb.StoreServiceManager, metrics
 		cacheSize:                cacheSize,
 		freshnessInterval:        freshnessInterval,
 		maxQueries:               maxQueries,
+		notifierDisabled:         notifierDisabled,
 	}
 }
 
@@ -88,7 +90,7 @@ func (p *fetcherProvider) GetFetcher(tmsID token.TMSID) (TokenFetcher, error) {
 		return nil, err
 	}
 
-	return p.fetch(tokenDB, p.metrics, p.cacheSize, p.freshnessInterval, p.maxQueries), nil
+	return p.fetch(tokenDB, p.metrics, p.cacheSize, p.freshnessInterval, p.maxQueries, p.notifierDisabled), nil
 }
 
 // mixedFetcher combines both eager and lazy strategies
@@ -149,14 +151,9 @@ func (f *lazyFetcher) UnspentTokensIteratorBy(ctx context.Context, walletID stri
 	return collections.NewPermutatedIterator[token2.UnspentTokenInWallet](it)
 }
 
-type permutatableIterator[T any] interface {
-	iterators.Iterator[T]
-	NewPermutation() iterators.Iterator[T]
-}
-
 type tokenCache interface {
-	Get(key string) (permutatableIterator[*token2.UnspentTokenInWallet], bool)
-	Add(key string, value permutatableIterator[*token2.UnspentTokenInWallet])
+	Get(key string) ([]*token2.UnspentTokenInWallet, bool)
+	Add(key string, value []*token2.UnspentTokenInWallet)
 	Delete(key string)
 	Clear()
 }
@@ -198,9 +195,9 @@ func NewCachedFetcher(tokenDB TokenDB, notifier dbdriver.TokenNotifier, cacheSiz
 	// If cacheSize <= 0, use default size; otherwise use custom size
 	// Both use the same default NumCounters and BufferItems
 	if cacheSize <= 0 {
-		ristrettoCache, err = cache.NewDefaultRistrettoCache[permutatableIterator[*token2.UnspentTokenInWallet]]()
+		ristrettoCache, err = utilcache.NewDefaultRistrettoCache[[]*token2.UnspentTokenInWallet]()
 	} else {
-		ristrettoCache, err = cache.NewRistrettoCacheWithSize[permutatableIterator[*token2.UnspentTokenInWallet]](cacheSize)
+		ristrettoCache, err = utilcache.NewRistrettoCacheWithSize[[]*token2.UnspentTokenInWallet](cacheSize)
 	}
 
 	if err != nil {
@@ -225,9 +222,54 @@ func NewCachedFetcher(tokenDB TokenDB, notifier dbdriver.TokenNotifier, cacheSiz
 }
 
 // onTokenChange is the callback registered with the token DB notifier.
-// Any write to the token table marks the cache dirty so the next query forces a refresh.
-func (f *cachedFetcher) onTokenChange(_ dbdriver.Operation, _ dbdriver.TokenRecordReference) {
-	f.dirty.Store(1)
+// When the reference carries wallet/type data it surgically patches the cache
+// bucket so the next query sees the change without a full DB scan.
+// If data is missing (empty WalletID or Type) we fall back to marking the
+// whole cache dirty so the next query triggers a full refresh.
+func (f *cachedFetcher) onTokenChange(op dbdriver.Operation, ref dbdriver.TokenRecordReference) {
+	if ref.WalletID == "" || ref.Type == "" {
+		f.dirty.Store(1)
+		return
+	}
+	key := tokenKey(ref.WalletID, ref.Type)
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	switch op {
+	case dbdriver.Insert:
+		toks, _ := f.cache.Get(key)
+		newTok := &token2.UnspentTokenInWallet{
+			Id:       token2.ID{TxId: ref.TxID, Index: ref.Index},
+			WalletID: ref.WalletID,
+			Type:     ref.Type,
+			Quantity: ref.Quantity,
+		}
+		f.cache.Add(key, append(toks, newTok))
+		f.prevKeys[key] = struct{}{}
+
+	case dbdriver.Delete:
+		toks, ok := f.cache.Get(key)
+		if !ok {
+			return
+		}
+		updated := toks[:0:0]
+		for _, t := range toks {
+			if t.Id.TxId != ref.TxID || t.Id.Index != ref.Index {
+				updated = append(updated, t)
+			}
+		}
+		if len(updated) == 0 {
+			f.cache.Delete(key)
+			delete(f.prevKeys, key)
+		} else {
+			f.cache.Add(key, updated)
+		}
+
+	default:
+		// For updates (e.g. spendable flag toggles) fall back to a full refresh.
+		f.dirty.Store(1)
+	}
 }
 
 func (f *cachedFetcher) update(ctx context.Context) {
@@ -278,7 +320,7 @@ func (f *cachedFetcher) updateCache(ctx context.Context, tokensByKey map[string]
 	// Step 1: Add/update new entries first
 	newKeys := make(map[string]struct{}, len(tokensByKey))
 	for key, toks := range tokensByKey {
-		f.cache.Add(key, iterators.Slice(toks))
+		f.cache.Add(key, toks)
 		newKeys[key] = struct{}{}
 	}
 
@@ -310,10 +352,10 @@ func (f *cachedFetcher) UnspentTokensIteratorBy(ctx context.Context, walletID st
 		f.mu.RLock()
 	}
 
-	it, ok := f.cache.Get(tokenKey(walletID, currency))
+	toks, ok := f.cache.Get(tokenKey(walletID, currency))
 	f.mu.RUnlock()
 	if ok {
-		return it.NewPermutation(), nil
+		return iterators.Slice(toks).NewPermutation(), nil
 	}
 	logger.DebugfContext(ctx, "No tokens found in cache for [%s]. Returning empty iterator.", tokenKey(walletID, currency))
 

token/services/selector/sherdlock/fetcher_test.go

@@ -327,7 +327,7 @@ func TestNewMixedFetcher(t *testing.T) {
 
 func TestRistrettoCache_Integration(t *testing.T) {
 	// Test that Ristretto cache works correctly with the fetcher
-	c, err := cache.NewRistrettoCacheWithSize[permutatableIterator[*token2.UnspentTokenInWallet]](10)
+	c, err := cache.NewRistrettoCacheWithSize[[]*token2.UnspentTokenInWallet](10)
 	require.NoError(t, err)
 	assert.NotNil(t, c)
 
@@ -339,8 +339,7 @@ func TestRistrettoCache_Integration(t *testing.T) {
 			Quantity: "100",
 		},
 	}
-	it := iterators.Slice(tokens)
-	c.Add("key1", it)
+	c.Add("key1", tokens)
 
 	// Wait for cache to process the addition (Ristretto is async)
 	time.Sleep(50 * time.Millisecond)
@@ -365,7 +364,7 @@ func TestRistrettoCache_Integration(t *testing.T) {
 func TestRistrettoCache_SizeLimit(t *testing.T) {
 	// Test that cache respects size limit
 	smallSize := int64(5)
-	c, err := cache.NewRistrettoCacheWithSize[permutatableIterator[*token2.UnspentTokenInWallet]](smallSize)
+	c, err := cache.NewRistrettoCacheWithSize[[]*token2.UnspentTokenInWallet](smallSize)
 	require.NoError(t, err)
 
 	// Add items with cost=1 each
@@ -377,8 +376,7 @@ func TestRistrettoCache_SizeLimit(t *testing.T) {
 				Quantity: "100",
 			},
 		}
-		it := iterators.Slice(tokens)
-		c.Add(tokenKey("wallet", token2.Type(string([]rune{rune(i)}))), it)
+		c.Add(tokenKey("wallet", token2.Type(string([]rune{rune(i)}))), tokens)
 	}
 
 	// Wait for cache to process additions
@@ -393,8 +391,7 @@ func TestRistrettoCache_SizeLimit(t *testing.T) {
 			Quantity: "100",
 		},
 	}
-	it := iterators.Slice(tokens)
-	c.Add("test_key", it)
+	c.Add("test_key", tokens)
 
 	// Wait for cache to process the addition
 	time.Sleep(50 * time.Millisecond)
@@ -798,6 +795,7 @@ func TestNewFetcherProvider(t *testing.T) {
 			100,
 			time.Second,
 			10,
+			false,
 		)
 
 		assert.NotNil(t, provider)
@@ -815,6 +813,7 @@ func TestNewFetcherProvider(t *testing.T) {
 				100,
 				time.Second,
 				10,
+				false,
 			)
 		})
 	})
@@ -827,6 +826,7 @@ func TestNewFetcherProvider(t *testing.T) {
 			0,
 			0,
 			0,
+			false,
 		)
 
 		assert.NotNil(t, provider)
@@ -852,6 +852,7 @@ func TestFetcherProvider_GetFetcher(t *testing.T) {
 			100,
 			time.Second,
 			10,
+			false,
 		)
 
 		fetcher, err := provider.GetFetcher(token.TMSID{})
@@ -875,6 +876,7 @@ func TestFetcherProvider_GetFetcher(t *testing.T) {
 			100,
 			time.Second,
 			10,
+			false,
 		)
 
 		fetcher, err := provider.GetFetcher(token.TMSID{})

token/services/selector/sherdlock/fetcher_unit_test.go

@@ -21,7 +21,7 @@ func TestFetcherProviderUnit(t *testing.T) {
 	mockSSM := &mocks.FakeTokenDBStoreServiceManager{}
 	metricsProvider, _ := setupMetricsMocks()
 
-	provider := sherdlock.NewFetcherProvider(mockSSM, metricsProvider, sherdlock.Mixed, 0, 0, 0)
+	provider := sherdlock.NewFetcherProvider(mockSSM, metricsProvider, sherdlock.Mixed, 0, 0, 0, false)
 
 	t.Run("GetFetcher_Error", func(t *testing.T) {
 		mockSSM.StoreServiceByTMSIdReturns(nil, errors.New("ssm error"))