Merge branch 'main' into docs/introduce-mkdocs

Author: SurbhiAgarwal1

token/core/zkatdlog/nogh/v1/crypto/rp/bulletproof/ipa.go

@@ -184,25 +184,63 @@ func (p *ipaProver) Prove() (*IPA, error) {
 // of the left vector and right is a function of right vector.
 // Both vectors are committed in com which is passed as a parameter to reduce
 func (p *ipaProver) reduce(X, com *mathlib.G1) (*mathlib.Zr, *mathlib.Zr, []*mathlib.G1, []*mathlib.G1, error) {
-	leftGen, rightGen := CloneGenerators(p.LeftGenerators, p.RightGenerators)
-
 	left := p.leftVector
 	right := p.rightVector
 
 	LArray := make([]*mathlib.G1, p.NumberOfRounds)
 	RArray := make([]*mathlib.G1, p.NumberOfRounds)
+	xList := make([]*mathlib.Zr, 0, p.NumberOfRounds)
+
 	for i := range p.NumberOfRounds {
 		// in each round the size of the vector is reduced by 2
-		n := len(leftGen) / 2
-		leftIP := math.InnerProduct(left[:n], right[n:], p.Curve)
-		rightIP := math.InnerProduct(left[n:], right[:n], p.Curve)
-		// LArray[i] is a commitment to left[:n], right[n:] and their inner product
-		LArray[i] = CommitVectorPlusOne(left[:n], right[n:], leftGen[n:], rightGen[:n], leftIP, X, p.Curve)
-		// LArray[i].Add(X.Mul(leftIP))
+		n_current := len(left) / 2
+		leftIP := math.InnerProduct(left[:n_current], right[n_current:], p.Curve)
+		rightIP := math.InnerProduct(left[n_current:], right[:n_current], p.Curve)
+
+		var s, sInv []*mathlib.Zr
+		if i == 0 {
+			s = []*mathlib.Zr{math.One(p.Curve)}
+			sInv = []*mathlib.Zr{math.One(p.Curve)}
+		} else {
+			s, sInv = ComputeSVector(1<<i, xList, p.Curve)
+		}
+
+		pointsL := make([]*mathlib.G1, 0, len(p.LeftGenerators)+1)
+		scalarsL := make([]*mathlib.Zr, 0, len(p.LeftGenerators)+1)
+
+		pointsR := make([]*mathlib.G1, 0, len(p.LeftGenerators)+1)
+		scalarsR := make([]*mathlib.Zr, 0, len(p.LeftGenerators)+1)
+
+		for m := range 1 << i {
+			for j := range n_current {
+				idxG_R := j + (2*m+1)*n_current
+				idxH_L := j + 2*m*n_current
+
+				pointsL = append(pointsL, p.LeftGenerators[idxG_R], p.RightGenerators[idxH_L])
+				scalarsL = append(scalarsL,
+					p.Curve.ModMul(left[j], s[m], p.Curve.GroupOrder),
+					p.Curve.ModMul(right[n_current+j], sInv[m], p.Curve.GroupOrder),
+				)
+
+				idxG_L := j + 2*m*n_current
+				idxH_R := j + (2*m+1)*n_current
+
+				pointsR = append(pointsR, p.LeftGenerators[idxG_L], p.RightGenerators[idxH_R])
+				scalarsR = append(scalarsR,
+					p.Curve.ModMul(left[n_current+j], s[m], p.Curve.GroupOrder),
+					p.Curve.ModMul(right[j], sInv[m], p.Curve.GroupOrder),
+				)
+			}
+		}
+
+		pointsL = append(pointsL, X)
+		scalarsL = append(scalarsL, leftIP)
+
+		pointsR = append(pointsR, X)
+		scalarsR = append(scalarsR, rightIP)
 
-		// RArray[i] is a commitment to left[n:], right[:n] and their inner product
-		RArray[i] = CommitVectorPlusOne(left[n:], right[:n], leftGen[:n], rightGen[n:], rightIP, X, p.Curve)
-		// RArray[i].Add(X.Mul(rightIP))
+		LArray[i] = p.Curve.MultiScalarMul(pointsL, scalarsL)
+		RArray[i] = p.Curve.MultiScalarMul(pointsR, scalarsR)
 
 		// compute this round's challenge x
 		array := common.GetG1Array([]*mathlib.G1{LArray[i], RArray[i]})
@@ -211,14 +249,12 @@ func (p *ipaProver) reduce(X, com *mathlib.G1) (*mathlib.Zr, *mathlib.Zr, []*mat
 			return nil, nil, nil, nil, err
 		}
 		x := p.Curve.HashToZr(bytesToHash)
+		xList = append(xList, x)
 
 		// compute 1/x
 		xInv := x.Copy()
 		xInv.InvModOrder()
 
-		// reduce the generators by 1/2, as a function of the old generators and x and 1/x
-		leftGen, rightGen = reduceGenerators(leftGen, rightGen, x, xInv, p.Provider)
-
 		// reduce the vectors by 1/2, a function of the old vectors and x and 1/x
 		left, right = reduceVectors(left, right, x, xInv, p.Curve)
 
@@ -404,27 +440,6 @@ func reduceVectors(left, right []*mathlib.Zr, x, xInv *mathlib.Zr, c *mathlib.Cu
 	return leftPrime, rightPrime
 }
 
-// reduceGenerators reduces the number of generators passed in the parameters by 1/2,
-// as a function of the old generators,  x and 1/x
-func reduceGenerators(leftGen, rightGen []*mathlib.G1, x, xInv *mathlib.Zr, provider executor.ExecutorProvider) ([]*mathlib.G1, []*mathlib.G1) {
-	l := len(leftGen) / 2
-	// Use the Executor abstraction so that the execution strategy can be
-	// swapped without changing this function. SerialExecutor runs each task
-	// immediately with no locks or goroutine overhead.
-	exec := provider.New()
-	for i := range l {
-		exec.Submit(func() {
-			// G_i = G_i^{x_inv} * G_{i+l}^x
-			leftGen[i].Mul2InPlace(xInv, leftGen[i+l], x)
-			// H_i = H_i^x * H_{i+l}^{x_inv}
-			rightGen[i].Mul2InPlace(x, rightGen[i+l], xInv)
-		})
-	}
-	exec.Wait()
-
-	return leftGen[:l], rightGen[:l]
-}
-
 func CommitVector(
 	left []*mathlib.Zr,
 	right []*mathlib.Zr,

token/services/auditor/mock/audit_transaction_store.go

@@ -618,6 +618,6 @@ type transactionDB interface {
 	GetTokenRequest(ctx context.Context, txID string) ([]byte, error)
 	SetStatus(ctx context.Context, txID string, status storage.TxStatus, message string) error
 	AcquireRecoveryLeadership(ctx context.Context, lockID int64) (recovery.Leadership, bool, error)
-	ClaimPendingTransactions(ctx context.Context, olderThan time.Duration, leaseDuration time.Duration, limit int, owner string) ([]*ttxdb.TransactionRecord, error)
+	ClaimPendingTransactions(ctx context.Context, olderThan time.Duration, leaseDuration time.Duration, limit int, owner string) ([]*ttxdb.RecoveryClaim, error)
 	ReleaseRecoveryClaim(ctx context.Context, txID string, owner string, message string) error
 }

token/services/network/fabric/network.go

@@ -8,10 +8,12 @@ package sherdlock
 
 import (
 	"context"
+	"io"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators"
 	"github.com/hyperledger-labs/fabric-token-sdk/token"
@@ -170,7 +172,11 @@ type cachedFetcher struct {
 	queriesResponded uint32
 	// prevKeys tracks cache keys from the previous update cycle to identify stale entries that need removal.
 	prevKeys map[string]struct{}
-	mu       sync.RWMutex
+	// isUpdating indicates if a cache refresh is currently in progress.
+	isUpdating bool
+	// updateCond allows goroutines to wait for an in-progress update to complete.
+	updateCond *sync.Cond
+	mu         sync.RWMutex
 }
 
 // NewCachedFetcher creates a fetcher that maintains a periodically refreshed cache of all tokens.
@@ -198,48 +204,107 @@ func NewCachedFetcher(tokenDB TokenDB, cacheSize int64, freshnessInterval time.D
 		panic("failed to create ristretto cache: " + err.Error())
 	}
 
-	return &cachedFetcher{
+	f := &cachedFetcher{
 		tokenDB:                 tokenDB,
 		cache:                   ristrettoCache,
 		freshnessInterval:       freshnessInterval,
 		maxQueriesBeforeRefresh: uint32(maxQueriesBeforeRefresh),
 		prevKeys:                make(map[string]struct{}),
 	}
+	f.updateCond = sync.NewCond(&f.mu)
+
+	return f
+}
+
+// finishUpdate releases the update lock and signals all waiting goroutines.
+// finishUpdate cleans up after an update operation: marks updating as complete,
+// broadcasts to waiting goroutines, and releases the lock.
+// Must be called while holding f.mu.
+func (f *cachedFetcher) finishUpdate() {
+	f.isUpdating = false
+	f.updateCond.Broadcast()
+	f.mu.Unlock()
 }
 
+// update refreshes the token cache from the database. It releases the lock during the
+// potentially slow DB operation to avoid blocking other goroutines, then re-acquires
+// the lock to atomically update the cache. A re-check of staleness is performed
+// after the DB call completes to avoid overwriting a cache that was refreshed by
+// another goroutine while waiting for the database.
 func (f *cachedFetcher) update(ctx context.Context) {
 	f.mu.Lock()
-	defer f.mu.Unlock()
+	if f.isUpdating {
+		// Wait for the in-progress update to finish
+		for f.isUpdating {
+			f.updateCond.Wait()
+		}
+		f.mu.Unlock()
+
+		return
+	}
+
 	if !f.isCacheStale() && !f.isCacheOverused() {
 		logger.DebugfContext(ctx, "Cache renewed in the meantime by another process")
+		f.mu.Unlock()
 
 		return
 	}
 	logger.DebugfContext(ctx, "Renew token cache")
+	f.isUpdating = true
+
+	// Release lock during slow DB operation to not block other token operations
+	f.mu.Unlock()
+
 	it, err := f.tokenDB.SpendableTokensIteratorBy(ctx, "", "")
 	if err != nil {
 		logger.Warnf("Failed to get token iterator: %v", err)
+		f.mu.Lock()
+		f.finishUpdate()
 
 		return
 	}
 	defer it.Close()
 
-	m := f.groupTokensByKey(ctx, it)
+	m, err := f.groupTokensByKey(ctx, it)
+	if err != nil {
+		logger.Warnf("Failed to group tokens from iterator: %v", err)
+		f.mu.Lock()
+		f.finishUpdate()
+
+		return
+	}
+
+	f.mu.Lock()
+	// Re-check: another goroutine may have refreshed while we waited for DB
+	if !f.isCacheStale() && !f.isCacheOverused() {
+		logger.DebugfContext(ctx, "Cache renewed in the meantime by another process, skipping")
+		f.finishUpdate()
+
+		return
+	}
+
 	f.updateCache(ctx, m)
 	atomic.StoreInt64(&f.lastFetched, time.Now().UnixNano())
 	atomic.StoreUint32(&f.queriesResponded, 0)
+	f.finishUpdate()
 }
 
 // groupTokensByKey reads tokens from the iterator and groups them by wallet/currency key.
-func (f *cachedFetcher) groupTokensByKey(ctx context.Context, it driver.SpendableTokensIterator) map[string][]*token2.UnspentTokenInWallet {
+// It returns an error if the iterator fails mid-way to prevent partial updates.
+func (f *cachedFetcher) groupTokensByKey(ctx context.Context, it driver.SpendableTokensIterator) (map[string][]*token2.UnspentTokenInWallet, error) {
 	m := map[string][]*token2.UnspentTokenInWallet{}
 	for t, err := it.Next(); err == nil && t != nil; t, err = it.Next() {
 		key := tokenKey(t.WalletID, t.Type)
 		logger.DebugfContext(ctx, "Adding token with key [%s]", key)
 		m[key] = append(m[key], t)
 	}
+	// Re-check for error after loop termination
+	_, err := it.Next()
+	if err != nil && !errors.Is(err, io.EOF) {
+		return nil, err
+	}
 
-	return m
+	return m, nil
 }
 
 // updateCache updates the cache by adding new entries before removing stale ones.