fix(identity): make RecipientDataCache provisioning cancellable

Signed-off-by: AkramBitar <akram@il.ibm.com>

Author: AkramBitar

.golangci.yml

@@ -106,6 +106,7 @@ linters:
         - go.opentelemetry.io/otel/trace.Span
         - github.com/hyperledger/fabric-lib-go/common/metrics.Gauge
         - github.com/hyperledger/fabric-lib-go/common/metrics.Histogram
+        - github.com/hyperledger-labs/fabric-smart-client/platform/view/services/metrics.Gauge
         - github.com/hyperledger-labs/fabric-smart-client/platform/common/driver.ConfigService
         - github.com/hyperledger-labs/fabric-smart-client/integration/nwo/api.ViewClient
         - github.com/hyperledger-labs/fabric-smart-client/integration/nwo/api.Platform

docs/services/identity.md

@@ -162,6 +162,74 @@ func (d *Base) NewWalletService(...) (*wallet.Service, error) {
 
 > **Note:** `provider` here is a `NewTMSProvider`-wrapped `Provider` (see [Driver Metrics](../drivers/metrics.md#pitfall-labelnames-must-include-network-channel-namespace)), which binds `network`/`channel`/`namespace` on every metric via `.With(...)` before returning it. Every `CounterOpts`/`HistogramOpts` above must therefore declare those three as `LabelNames` in addition to its own label(s), or the metric panics with "inconsistent label cardinality" on first use. This is exactly the bug that crashed the DVP/DLog integration suite in `SignerRouter.Register` before it was fixed.
 
+## Wallet Lifecycle and Recipient Data Caching
+
+Anonymous owner wallets hand out a fresh pseudonym for every payment. Generating one is
+expensive (an Idemix pseudonym plus a registry binding), so `AnonymousOwnerWallet` keeps a
+pre-provisioned buffer of recipient data. Two caches implement that buffer:
+
+| Cache | Buffers | Sized by |
+|:------|:--------|:---------|
+| `role.RecipientDataCache` (`token/services/identity/role/cache.go`) | `driver.RecipientData` (pseudonym + audit info) for one wallet | `wallets.owners[].cacheSize`, falling back to `wallets.defaultCacheSize` (see [configuration](../configuration.md)) |
+| `idemix/cache.IdentityCache` (`token/services/identity/idemix/cache/cache.go`) | `idriver.IdentityDescriptor` for one Idemix key manager | same lookup, via `KeyManagerProvider.cacheSizeForID` |
+
+Both follow the same contract:
+
+*   **Provisioning is lazy.** The background goroutine is started by the first request, and
+    only when the configured size is greater than zero. With a size of zero the cache is
+    disabled and every request goes straight to the backend.
+*   **Requests never wait on the cache.** A request that does not find a buffered entry
+    within a short timeout (5 ms) generates the data on the spot instead of blocking, so a
+    slow backend degrades latency rather than stalling the caller. A cancelled caller
+    context aborts the request immediately.
+*   **A failing backend backs off, and is observable.** The provisioning loop logs the
+    failure, increments a counter and waits one second before retrying, so a broken
+    identity backend cannot turn pre-provisioning into a busy loop — and the condition can
+    be alerted on instead of only appearing in the logs.
+*   **`Close()` is mandatory and idempotent.** It cancels the background context, which
+    terminates the provisioning goroutine even while it is parked on a full buffer or
+    inside a retry backoff. A cache that is never closed keeps its goroutine, its channel
+    and its backend closure alive for the lifetime of the process. After `Close()` the
+    cache still serves requests from the backend; it simply stops pre-provisioning.
+
+### Cache metrics
+
+| Metric | Type | Cache | Purpose |
+|:-------|:-----|:------|:--------|
+| `recipient_data_cache_level` | Gauge | `RecipientDataCache` | Entries currently buffered. Counted only once an entry is really in the buffer, so it cannot drift upward when the producer is blocked. |
+| `recipient_data_provision_failures_total` | Counter | `RecipientDataCache` | Failed pre-provisioning attempts. A rising rate means the identity backend is failing and requests are falling back to the slower on-demand path. |
+| `cache_level` | Gauge | idemix `IdentityCache` | As above, for Idemix identities. |
+| `cache_provision_failures_total` | Counter | idemix `IdentityCache` | As above, for Idemix identities. |
+
+> **Note:** these providers are `NewTMSProvider`-wrapped, so every `GaugeOpts`/`CounterOpts`
+> above must declare `network`, `channel` and `namespace` in `LabelNames` — omitting them
+> panics with "inconsistent label cardinality" on first use. See
+> [Driver Metrics](../drivers/metrics.md#pitfall-labelnames-must-include-network-channel-namespace).
+
+### Who calls `Close()`
+
+Application code does not normally close these caches itself: they are released by the
+existing teardown chain when a token management service is unloaded, for instance when
+its public parameters are updated.
+
+```
+core.TMSProvider.Update            (token/core/tms.go)
+  └── Service.Done()               (token/core/common/tms.go)
+        └── wallet.Service.Done()  (token/services/identity/wallet/service.go)
+              └── role.Registry.Done()
+                    ├── Close() on every wallet it created that holds resources
+                    │     └── AnonymousOwnerWallet.Close() → RecipientDataCache.Close()
+                    └── Role.Done() → LocalMembership.Close()
+```
+
+`role.Registry.Done()` closes wallets through a local `interface{ Close() }` assertion
+rather than through `driver.Wallet`, so wallet types with nothing to release need not
+implement a no-op `Close()`. If you add a wallet type that owns a goroutine, a ticker or
+any other resource, give it a `Close()` method and it will be released automatically.
+
+> **Note:** tests that exercise an anonymous owner wallet should `t.Cleanup(w.Close)`,
+> otherwise each test leaves a provisioning goroutine behind for the rest of the run.
+
 ## Identity Types
 
 The Identity Service leverages a wrapper called **TypedIdentity** to support various identity schemes uniformly. 

token/services/identity/idemix/cache/cache.go

@@ -36,29 +36,37 @@ type IdentityCache struct {
 	cacheTimeout time.Duration
 	// Cache performance metrics
 	metrics *Metrics
-	// Cancellation function for background provisioning
+	// provisionCtx governs the lifetime of the background provisioning goroutine.
+	// Created at construction time and cancelled by Close.
+	provisionCtx context.Context
+	// cancel cancels provisionCtx. Written once, at construction time, so that Close
+	// is safe to call concurrently with Identity.
 	cancel context.CancelFunc
 }
 
 // NewIdentityCache creates a new identity cache with specified size and backend.
 func NewIdentityCache(backed IdentityCacheBackendFunc, size int, auditInfo []byte, metrics *Metrics) *IdentityCache {
 	logger.Debugf("new identity cache with size [%d]", size)
+	// The provisioning goroutine must not inherit any caller's request context, but it
+	// must still be cancellable, hence a cancellable child of context.Background.
+	provisionCtx, cancel := context.WithCancel(context.Background())
 	ci := &IdentityCache{
 		backed:       backed,
 		cache:        make(chan *idriver.IdentityDescriptor, size),
 		auditInfo:    auditInfo,
 		cacheTimeout: 5 * time.Millisecond,
 		metrics:      metrics,
+		provisionCtx: provisionCtx,
+		cancel:       cancel,
 	}
 
 	return ci
 }
 
-// Close stops the background identity provisioning.
+// Close stops the background identity provisioning. It is idempotent and safe to call
+// even if provisioning was never started, or concurrently with Identity.
 func (c *IdentityCache) Close() {
-	if c.cancel != nil {
-		c.cancel()
-	}
+	c.cancel()
 }
 
 // Identity retrieves an identity from cache or generates on-demand.
@@ -70,10 +78,10 @@ func (c *IdentityCache) Identity(ctx context.Context, auditInfo []byte) (*idrive
 
 	c.once.Do(func() {
 		logger.DebugfContext(ctx, "provision identities with cache size [%d]", cap(c.cache))
-		if cap(c.cache) > 0 {
-			var backgroundCtx context.Context
-			backgroundCtx, c.cancel = context.WithCancel(context.Background())
-			go c.provisionIdentities(backgroundCtx)
+		// Do not spawn the goroutine if the cache has already been closed, otherwise a
+		// late first call would start a goroutine that nothing will ever stop.
+		if cap(c.cache) > 0 && c.provisionCtx.Err() == nil {
+			go c.provisionIdentities(c.provisionCtx)
 		}
 	})
 
@@ -143,6 +151,7 @@ func (c *IdentityCache) provisionIdentities(ctx context.Context) {
 	for {
 		identityDescriptor, err := c.backed(ctx, c.auditInfo)
 		if err != nil {
+			c.metrics.ProvisionFailuresCount.Add(1)
 			logger.Errorf("failed to provision identity [%s]", err)
 			select {
 			case <-ctx.Done():
@@ -153,9 +162,11 @@ func (c *IdentityCache) provisionIdentities(ctx context.Context) {
 			continue
 		}
 		logger.DebugfContext(ctx, "generated new idemix identity [%d]", count)
-		c.metrics.CacheLevelGauge.Add(1)
+		// The gauge is incremented only once the entry is actually in the channel, so a
+		// cancellation mid-send cannot leave the reported cache level skewed.
 		select {
 		case c.cache <- identityDescriptor:
+			c.metrics.CacheLevelGauge.Add(1)
 			count++
 		case <-ctx.Done():
 			return

token/services/identity/idemix/cache/cache_test.go

@@ -9,13 +9,16 @@ package cache
 import (
 	"context"
 	"errors"
+	"runtime"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/LFDT-Panurus/panurus/token/driver"
 	idriver "github.com/LFDT-Panurus/panurus/token/services/identity/driver"
+	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/metrics"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/metrics/disabled"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -29,6 +32,7 @@ func TestIdentityCache(t *testing.T) {
 			AuditInfo: []byte("audit"),
 		}, nil
 	}, 100, nil, NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 	identityDescriptor, err := c.Identity(t.Context(), nil)
 	require.NoError(t, err)
 	assert.Equal(t, driver.Identity([]byte("hello world")), identityDescriptor.Identity)
@@ -48,6 +52,7 @@ func TestIdentityCacheForRace(t *testing.T) {
 			AuditInfo: []byte("audit"),
 		}, nil
 	}, 10000, nil, NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 
 	numRoutines := 4
 	wg := sync.WaitGroup{}
@@ -77,6 +82,7 @@ func TestFetchIdentityFromBackend(t *testing.T) {
 	c := NewIdentityCache(func(ctx context.Context, auditInfo []byte) (*idriver.IdentityDescriptor, error) {
 		return expectedIdentity, nil
 	}, 10, []byte("cache audit"), NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 
 	// Call with different audit info to trigger backend fetch
 	identityDescriptor, err := c.Identity(context.Background(), []byte("different audit"))
@@ -92,6 +98,7 @@ func TestFetchIdentityFromBackendError(t *testing.T) {
 	c := NewIdentityCache(func(ctx context.Context, auditInfo []byte) (*idriver.IdentityDescriptor, error) {
 		return nil, expectedErr
 	}, 10, []byte("cache audit"), NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 
 	// Call with different audit info to trigger backend fetch
 	_, err := c.Identity(context.Background(), []byte("different audit"))
@@ -111,6 +118,7 @@ func TestFetchIdentityFromCacheTimeout(t *testing.T) {
 			AuditInfo: []byte("timeout audit"),
 		}, nil
 	}, 0, nil, NewMetrics(&disabled.Provider{})) // cache size 0 to force timeout
+	t.Cleanup(c.Close)
 
 	// Set short timeout to trigger timeout path
 	c.cacheTimeout = 1 * time.Millisecond
@@ -129,6 +137,7 @@ func TestFetchIdentityFromCacheTimeoutError(t *testing.T) {
 	c := NewIdentityCache(func(ctx context.Context, auditInfo []byte) (*idriver.IdentityDescriptor, error) {
 		return nil, expectedErr
 	}, 0, nil, NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 
 	// Set short timeout to trigger timeout path
 	c.cacheTimeout = 1 * time.Millisecond
@@ -183,6 +192,7 @@ func TestFetchIdentityFromCacheNilEntry(t *testing.T) {
 			AuditInfo: []byte("backend audit"),
 		}, nil
 	}, 10, nil, NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 
 	// Pre-populate the cache with nil before calling Identity()
 	// Since cache is buffered, this completes immediately
@@ -213,6 +223,7 @@ func TestIdentityCache_Close(t *testing.T) {
 	}
 
 	c := NewIdentityCache(backend, 10, nil, NewMetrics(&disabled.Provider{}))
+	t.Cleanup(c.Close)
 	// Set a very short timeout so we don't wait long if the cache is empty
 	c.cacheTimeout = 1 * time.Millisecond
 
@@ -239,3 +250,92 @@ func TestIdentityCache_Close(t *testing.T) {
 	// may complete a few in-flight iterations before observing the stop signal.
 	assert.LessOrEqual(t, callCount.Load(), countAfterClose+3)
 }
+
+// provisionGoroutineName is the symbol appearing in the stack trace of the background
+// provisioning goroutine.
+const provisionGoroutineName = "cache.(*IdentityCache).provisionIdentities"
+
+// provisioningGoroutines counts the running provisioning goroutines.
+func provisioningGoroutines() int {
+	buf := make([]byte, 1<<16)
+	for {
+		n := runtime.Stack(buf, true)
+		if n < len(buf) {
+			return strings.Count(string(buf[:n]), provisionGoroutineName)
+		}
+		buf = make([]byte, 2*len(buf))
+	}
+}
+
+// TestIdentityCache_CloseRacesFirstUse checks Close is safe to call concurrently with
+// the first Identity call, and that the cancellation is never missed.
+//
+// The cancel function used to be assigned inside once.Do, so Close read it without
+// synchronisation: a data race, and worse, a Close that observed the old nil value
+// silently skipped the cancellation and leaked the goroutine. Building the context in
+// the constructor makes the field write-once.
+func TestIdentityCache_CloseRacesFirstUse(t *testing.T) {
+	require.Eventually(t, func() bool {
+		return provisioningGoroutines() == 0
+	}, 5*time.Second, 10*time.Millisecond, "a previous test left a provisioning goroutine behind")
+
+	for range 300 {
+		c := NewIdentityCache(func(context.Context, []byte) (*idriver.IdentityDescriptor, error) {
+			return &idriver.IdentityDescriptor{Identity: []byte("id")}, nil
+		}, 4, nil, NewMetrics(&disabled.Provider{}))
+
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_, _ = c.Identity(context.Background(), nil)
+		}()
+		go func() {
+			defer wg.Done()
+			c.Close()
+		}()
+		wg.Wait()
+	}
+
+	// Every cache was closed, so no provisioning goroutine may survive.
+	require.Eventually(t, func() bool {
+		return provisioningGoroutines() == 0
+	}, 5*time.Second, 10*time.Millisecond, "Close missed the cancellation and leaked a goroutine")
+}
+
+// countingCounter is a minimal metrics.Counter that keeps a running total.
+type countingCounter struct {
+	total atomic.Int64
+}
+
+func (c *countingCounter) With(...string) metrics.Counter { return c }
+
+func (c *countingCounter) Add(delta float64) { c.total.Add(int64(delta)) }
+
+func (c *countingCounter) value() float64 { return float64(c.total.Load()) }
+
+// TestIdentityCache_CountsProvisionFailures checks a failing key manager is reported on a
+// counter, not only in the log, so the condition can be alerted on.
+func TestIdentityCache_CountsProvisionFailures(t *testing.T) {
+	failures := &countingCounter{}
+	c := NewIdentityCache(func(context.Context, []byte) (*idriver.IdentityDescriptor, error) {
+		return nil, errors.New("key manager is down")
+	}, 10, nil, &Metrics{CacheLevelGauge: &noopGauge{}, ProvisionFailuresCount: failures})
+	t.Cleanup(c.Close)
+
+	_, err := c.Identity(context.Background(), nil)
+	require.Error(t, err)
+
+	require.Eventually(t, func() bool {
+		return failures.value() >= 1
+	}, 2*time.Second, 5*time.Millisecond, "the failed provisioning attempt was not counted")
+}
+
+// noopGauge is a metrics.Gauge that discards everything.
+type noopGauge struct{}
+
+func (g *noopGauge) With(...string) metrics.Gauge { return g }
+
+func (g *noopGauge) Add(float64) {}
+
+func (g *noopGauge) Set(float64) {}

token/services/identity/idemix/cache/metrics.go

@@ -17,17 +17,34 @@ var (
 		Help:       "Level of the idemix cache",
 		LabelNames: []string{"network", "channel", "namespace"},
 	}
+
+	// ProvisionFailuresOpts counts failed attempts to pre-provision identities. A
+	// rising rate means the key manager is failing: identities are generated on the
+	// request path instead and the background loop is retrying. This is the signal to
+	// alert on, since the corresponding log line alone is easy to miss.
+	//
+	// LabelNames must repeat network/channel/namespace: the provider is TMS-wrapped
+	// and binds those three on every metric, so omitting them here panics with
+	// "inconsistent label cardinality" on first use.
+	ProvisionFailuresOpts = metrics.CounterOpts{
+		Name:       "cache_provision_failures_total",
+		Help:       "Failed attempts to pre-provision idemix identities",
+		LabelNames: []string{"network", "channel", "namespace"},
+	}
 )
 
 // Metrics contains metrics for monitoring identity cache performance.
 type Metrics struct {
 	// Current number of cached identities
 	CacheLevelGauge metrics.Gauge
+	// Failed background provisioning attempts
+	ProvisionFailuresCount metrics.Counter
 }
 
 // NewMetrics creates a new Metrics instance.
 func NewMetrics(p metrics.Provider) *Metrics {
 	return &Metrics{
-		CacheLevelGauge: p.NewGauge(LevelOpts),
+		CacheLevelGauge:        p.NewGauge(LevelOpts),
+		ProvisionFailuresCount: p.NewCounter(ProvisionFailuresOpts),
 	}
 }