Harden RPC-fault reliability: bound locks/timeouts and cap fan-out (7 fixes)

decentralized-api/internal/authzcache/cache.go

@@ -8,10 +8,17 @@ import (
 	"time"
 
 	"github.com/productscience/inference/x/inference/types"
+	"golang.org/x/sync/singleflight"
 )
 
 const authzCacheTTL = 2 * time.Minute
 
+// authzQueryTimeout bounds the coalesced signer lookup. The queries run under a
+// fresh context, not any single caller's, so one coalesced caller cancelling
+// (client disconnect or its own deadline firing) does not fail every other
+// caller that shares the singleflight result.
+const authzQueryTimeout = 15 * time.Second
+
 // SignerInfo holds address and pubkey for an authorized signer.
 type SignerInfo struct {
 	Address string
@@ -26,8 +33,13 @@ type cachedEntry struct {
 // AuthzCache caches authorized signers for granter addresses to avoid repeated chain queries.
 // Keys are cached with TTL since authz grants can change.
 type AuthzCache struct {
-	mu       sync.RWMutex
-	cache    map[string]*cachedEntry // "granterAddress|msgTypeUrl" -> entry
+	mu    sync.RWMutex
+	cache map[string]*cachedEntry // "granterAddress|msgTypeUrl" -> entry
+	// sf coalesces concurrent misses for the same key into a single pair of chain
+	// queries, so an epoch-boundary/expiry burst does not fan out N identical
+	// lookups; combined with running the queries outside c.mu it stops one slow
+	// granter from throttling verification for every other granter.
+	sf       singleflight.Group
 	recorder cosmosclient.CosmosMessageClient
 }
 
@@ -77,6 +89,7 @@ func (c *AuthzCache) GetPubKeyForSigner(ctx context.Context, granterAddress, sig
 func (c *AuthzCache) getSigners(ctx context.Context, granterAddress, msgTypeUrl string) ([]SignerInfo, error) {
 	cacheKey := granterAddress + "|" + msgTypeUrl
 
+	// Fast path: cache hit under the read lock.
 	c.mu.RLock()
 	if entry, ok := c.cache[cacheKey]; ok && time.Now().Before(entry.expiresAt) {
 		signers := entry.signers
@@ -85,56 +98,81 @@ func (c *AuthzCache) getSigners(ctx context.Context, granterAddress, msgTypeUrl
 	}
 	c.mu.RUnlock()
 
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	// Miss: coalesce concurrent fetches for this key into one pair of RPCs, run
+	// OUTSIDE c.mu so a slow granter cannot throttle verification for everyone.
+	// DoChan (not Do) so this caller can still abandon its wait via its own ctx
+	// without killing the shared query for everyone else.
+	ch := c.sf.DoChan(cacheKey, func() (interface{}, error) {
+		// Re-check: another goroutine may have populated the cache while we were
+		// becoming the singleflight leader.
+		c.mu.RLock()
+		if entry, ok := c.cache[cacheKey]; ok && time.Now().Before(entry.expiresAt) {
+			signers := entry.signers
+			c.mu.RUnlock()
+			return signers, nil
+		}
+		c.mu.RUnlock()
 
-	// Double-check after acquiring write lock
-	if entry, ok := c.cache[cacheKey]; ok && time.Now().Before(entry.expiresAt) {
-		return entry.signers, nil
-	}
+		logging.Debug("Fetching authz signers", types.Validation,
+			"granterAddress", granterAddress, "msgTypeUrl", msgTypeUrl)
 
-	logging.Debug("Fetching authz signers", types.Validation,
-		"granterAddress", granterAddress, "msgTypeUrl", msgTypeUrl)
+		queryClient := c.recorder.NewInferenceQueryClient()
 
-	queryClient := c.recorder.NewInferenceQueryClient()
+		// Run under a fresh, decoupled context so a coalesced caller's
+		// cancellation cannot fail everyone sharing this singleflight result.
+		qctx, cancel := context.WithTimeout(context.Background(), authzQueryTimeout)
+		defer cancel()
 
-	// Get grantees (warm keys) for this message type
-	grantees, err := queryClient.GranteesByMessageType(ctx, &types.QueryGranteesByMessageTypeRequest{
-		GranterAddress: granterAddress,
-		MessageTypeUrl: msgTypeUrl,
-	})
-	if err != nil {
-		return nil, err
-	}
+		// Get grantees (warm keys) for this message type
+		grantees, err := queryClient.GranteesByMessageType(qctx, &types.QueryGranteesByMessageTypeRequest{
+			GranterAddress: granterAddress,
+			MessageTypeUrl: msgTypeUrl,
+		})
+		if err != nil {
+			return nil, err
+		}
 
-	// Get granter's own public key
-	participant, err := queryClient.AccountByAddress(ctx, &types.QueryAccountByAddressRequest{
-		Address: granterAddress,
-	})
-	if err != nil {
-		return nil, err
-	}
+		// Get granter's own public key
+		participant, err := queryClient.AccountByAddress(qctx, &types.QueryAccountByAddressRequest{
+			Address: granterAddress,
+		})
+		if err != nil {
+			return nil, err
+		}
 
-	// Collect all signers: grantees + granter
-	signers := make([]SignerInfo, 0, len(grantees.Grantees)+1)
-	for _, grantee := range grantees.Grantees {
+		// Collect all signers: grantees + granter
+		signers := make([]SignerInfo, 0, len(grantees.Grantees)+1)
+		for _, grantee := range grantees.Grantees {
+			signers = append(signers, SignerInfo{
+				Address: grantee.Address,
+				PubKey:  grantee.PubKey,
+			})
+		}
 		signers = append(signers, SignerInfo{
-			Address: grantee.Address,
-			PubKey:  grantee.PubKey,
+			Address: granterAddress,
+			PubKey:  participant.Pubkey,
 		})
-	}
-	signers = append(signers, SignerInfo{
-		Address: granterAddress,
-		PubKey:  participant.Pubkey,
-	})
 
-	c.cache[cacheKey] = &cachedEntry{
-		signers:   signers,
-		expiresAt: time.Now().Add(authzCacheTTL),
-	}
+		c.mu.Lock()
+		c.cache[cacheKey] = &cachedEntry{
+			signers:   signers,
+			expiresAt: time.Now().Add(authzCacheTTL),
+		}
+		c.mu.Unlock()
+
+		logging.Debug("Cached authz signers", types.Validation,
+			"granterAddress", granterAddress, "count", len(signers))
 
-	logging.Debug("Cached authz signers", types.Validation,
-		"granterAddress", granterAddress, "count", len(signers))
+		return signers, nil
+	})
 
-	return signers, nil
+	select {
+	case res := <-ch:
+		if res.Err != nil {
+			return nil, res.Err
+		}
+		return res.Val.([]SignerInfo), nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
 }

decentralized-api/internal/authzcache/cache_fetch_test.go

@@ -0,0 +1,146 @@
+package authzcache
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"decentralized-api/cosmosclient"
+
+	"github.com/productscience/inference/x/inference/types"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+// fakeAuthzQueryClient implements only the two query methods getSigners uses; the
+// embedded (nil) interface satisfies the rest and panics if anything else runs.
+type fakeAuthzQueryClient struct {
+	types.QueryClient
+	granteesFn func(ctx context.Context) (*types.QueryGranteesByMessageTypeResponse, error)
+	accountFn  func(ctx context.Context) (*types.QueryAccountByAddressResponse, error)
+}
+
+func (f *fakeAuthzQueryClient) GranteesByMessageType(ctx context.Context, in *types.QueryGranteesByMessageTypeRequest, opts ...grpc.CallOption) (*types.QueryGranteesByMessageTypeResponse, error) {
+	return f.granteesFn(ctx)
+}
+
+func (f *fakeAuthzQueryClient) AccountByAddress(ctx context.Context, in *types.QueryAccountByAddressRequest, opts ...grpc.CallOption) (*types.QueryAccountByAddressResponse, error) {
+	return f.accountFn(ctx)
+}
+
+type fakeAuthzRecorder struct {
+	cosmosclient.CosmosMessageClient
+	qc types.QueryClient
+}
+
+func (r *fakeAuthzRecorder) NewInferenceQueryClient() types.QueryClient { return r.qc }
+
+// TestGetSignersSingleflightsAndReleasesLock proves that concurrent misses for
+// the same granter coalesce into one pair of chain queries, and that the cache
+// mutex is not held across the queries, so a slow granter cannot throttle
+// verification for every other granter.
+func TestGetSignersSingleflightsAndReleasesLock(t *testing.T) {
+	var granteesCalls int32
+	entered := make(chan struct{}, 1)
+	release := make(chan struct{})
+
+	qc := &fakeAuthzQueryClient{
+		granteesFn: func(ctx context.Context) (*types.QueryGranteesByMessageTypeResponse, error) {
+			atomic.AddInt32(&granteesCalls, 1)
+			select {
+			case entered <- struct{}{}:
+			default:
+			}
+			<-release
+			return &types.QueryGranteesByMessageTypeResponse{Grantees: []*types.Grantee{{Address: "g1", PubKey: "pk1"}}}, nil
+		},
+		accountFn: func(ctx context.Context) (*types.QueryAccountByAddressResponse, error) {
+			return &types.QueryAccountByAddressResponse{Pubkey: "granterpk"}, nil
+		},
+	}
+	cache := NewAuthzCache(&fakeAuthzRecorder{qc: qc})
+
+	const N = 8
+	var wg sync.WaitGroup
+	for i := 0; i < N; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, _ = cache.GetPubKeys(context.Background(), "granter", "msg")
+		}()
+	}
+
+	select {
+	case <-entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("no RPC started")
+	}
+
+	locked := make(chan struct{})
+	go func() {
+		cache.mu.RLock()
+		cache.mu.RUnlock()
+		close(locked)
+	}()
+	select {
+	case <-locked:
+	case <-time.After(time.Second):
+		t.Fatal("cache mutex held across the RPC - one slow granter would throttle everyone")
+	}
+
+	close(release)
+	wg.Wait()
+	require.Equal(t, int32(1), atomic.LoadInt32(&granteesCalls), "singleflight must coalesce concurrent misses into one query pair")
+}
+
+// TestGetPubKeyForSignerAndCaching covers correctness of the fetch path and that a
+// single fetch populates the cache for all subsequent lookups within the TTL.
+func TestGetPubKeyForSignerAndCaching(t *testing.T) {
+	var calls int32
+	qc := &fakeAuthzQueryClient{
+		granteesFn: func(ctx context.Context) (*types.QueryGranteesByMessageTypeResponse, error) {
+			atomic.AddInt32(&calls, 1)
+			return &types.QueryGranteesByMessageTypeResponse{Grantees: []*types.Grantee{{Address: "grantee1", PubKey: "gpk1"}}}, nil
+		},
+		accountFn: func(ctx context.Context) (*types.QueryAccountByAddressResponse, error) {
+			return &types.QueryAccountByAddressResponse{Pubkey: "granterpk"}, nil
+		},
+	}
+	cache := NewAuthzCache(&fakeAuthzRecorder{qc: qc})
+
+	pk, err := cache.GetPubKeyForSigner(context.Background(), "granter", "grantee1", "msg")
+	require.NoError(t, err)
+	require.Equal(t, "gpk1", pk)
+
+	pk, err = cache.GetPubKeyForSigner(context.Background(), "granter", "granter", "msg")
+	require.NoError(t, err)
+	require.Equal(t, "granterpk", pk)
+
+	pk, err = cache.GetPubKeyForSigner(context.Background(), "granter", "nobody", "msg")
+	require.NoError(t, err)
+	require.Equal(t, "", pk, "unknown signer returns empty, not an error")
+
+	pks, err := cache.GetPubKeys(context.Background(), "granter", "msg")
+	require.NoError(t, err)
+	require.ElementsMatch(t, []string{"gpk1", "granterpk"}, pks)
+
+	require.Equal(t, int32(1), atomic.LoadInt32(&calls), "all lookups served from a single cached fetch")
+}
+
+func TestGetSignersPropagatesRPCError(t *testing.T) {
+	qc := &fakeAuthzQueryClient{
+		granteesFn: func(ctx context.Context) (*types.QueryGranteesByMessageTypeResponse, error) {
+			return nil, fmt.Errorf("chain unavailable")
+		},
+		accountFn: func(ctx context.Context) (*types.QueryAccountByAddressResponse, error) {
+			return &types.QueryAccountByAddressResponse{}, nil
+		},
+	}
+	cache := NewAuthzCache(&fakeAuthzRecorder{qc: qc})
+
+	_, err := cache.GetPubKeys(context.Background(), "granter", "msg")
+	require.Error(t, err)
+}