[release-v2.0] main: Use backported mixing updates.

go.mod

@@ -26,7 +26,7 @@ require (
 	github.com/decred/dcrd/gcs/v4 v4.1.0
 	github.com/decred/dcrd/lru v1.1.2
 	github.com/decred/dcrd/math/uint256 v1.0.2
-	github.com/decred/dcrd/mixing v0.4.1
+	github.com/decred/dcrd/mixing v0.4.2
 	github.com/decred/dcrd/peer/v3 v3.1.3
 	github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.3.0
 	github.com/decred/dcrd/rpcclient/v8 v8.0.1
@@ -47,7 +47,7 @@ require (
 )
 
 require (
-	decred.org/cspp/v2 v2.2.0 // indirect
+	decred.org/cspp/v2 v2.3.0 // indirect
 	github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412 // indirect
 	github.com/companyzero/sntrup4591761 v0.0.0-20220309191932-9e0f3af2f07a // indirect
 	github.com/dchest/siphash v1.2.3 // indirect

go.sum

@@ -1,5 +1,5 @@
-decred.org/cspp/v2 v2.2.0 h1:VSOUC1w0Wo+QOGS0r1XO6TLnO16X67KuvpDmRRYyr08=
-decred.org/cspp/v2 v2.2.0/go.mod h1:9nO3bfvCheOPIFZw5f6sRQ42CjBFB5RKSaJ9Iq6G4MA=
+decred.org/cspp/v2 v2.3.0 h1:GC8emJnLbOVAkgBTHK/1wy6o/m0AVsN1r4m1ZnZZWjo=
+decred.org/cspp/v2 v2.3.0/go.mod h1:9nO3bfvCheOPIFZw5f6sRQ42CjBFB5RKSaJ9Iq6G4MA=
 github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412 h1:w1UutsfOrms1J05zt7ISrnJIXKzwaspym5BTKGx93EI=
 github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412/go.mod h1:WPjqKcmVOxf0XSf3YxCJs6N6AOSrOx3obionmG7T0y0=
 github.com/companyzero/sntrup4591761 v0.0.0-20220309191932-9e0f3af2f07a h1:clYxJ3Os0EQUKDDVU8M0oipllX0EkuFNBfhVQuIfyF0=
@@ -58,8 +58,8 @@ github.com/decred/dcrd/lru v1.1.2 h1:KdCzlkxppuoIDGEvCGah1fZRicrDH36IipvlB1ROkFY
 github.com/decred/dcrd/lru v1.1.2/go.mod h1:gEdCVgXs1/YoBvFWt7Scgknbhwik3FgVSzlnCcXL2N8=
 github.com/decred/dcrd/math/uint256 v1.0.2 h1:o8peafL5QmuXGTergI3YDpDU0eq5Z0pQi88B8ym4PRA=
 github.com/decred/dcrd/math/uint256 v1.0.2/go.mod h1:7M/y9wJJvlyNG/f/X6mxxhxo9dgloZHFiOfbiscl75A=
-github.com/decred/dcrd/mixing v0.4.1 h1:W8ZCzhmNyzG1xjJMA3L6FOElmp98Ttnk3dDUxD6irAE=
-github.com/decred/dcrd/mixing v0.4.1/go.mod h1:ySvVwTZyVz5YvevA6YjPrB6pJEwTm7IkHohTfaiHh2c=
+github.com/decred/dcrd/mixing v0.4.2 h1:mpt2pNIFTI6L1hXrieAWJTQJv5t9WzHcNnhI+tnAG90=
+github.com/decred/dcrd/mixing v0.4.2/go.mod h1:VF87lOn41kitgWVOwmXoB4qMYF7+bxItZXyw4JfW3EQ=
 github.com/decred/dcrd/peer/v3 v3.1.3 h1:MmIVuP82VdBTqSqqz6+m8YFpsV7C7KT/jTzngleepyY=
 github.com/decred/dcrd/peer/v3 v3.1.3/go.mod h1:Uj2o/et9DswIv173UTIiQwZ3jb57vcjss8yAkUHAvcY=
 github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.3.0 h1:l0DnCcILTNrpy8APF3FLN312ChpkQaAuW30aC/RgBaw=

mixing/go.mod

@@ -3,7 +3,7 @@ module github.com/decred/dcrd/mixing
 go 1.18
 
 require (
-	decred.org/cspp/v2 v2.2.0
+	decred.org/cspp/v2 v2.3.0
 	github.com/companyzero/sntrup4591761 v0.0.0-20220309191932-9e0f3af2f07a
 	github.com/davecgh/go-spew v1.1.1
 	github.com/decred/dcrd/chaincfg/chainhash v1.0.4

mixing/go.sum

@@ -1,5 +1,5 @@
-decred.org/cspp/v2 v2.2.0 h1:VSOUC1w0Wo+QOGS0r1XO6TLnO16X67KuvpDmRRYyr08=
-decred.org/cspp/v2 v2.2.0/go.mod h1:9nO3bfvCheOPIFZw5f6sRQ42CjBFB5RKSaJ9Iq6G4MA=
+decred.org/cspp/v2 v2.3.0 h1:GC8emJnLbOVAkgBTHK/1wy6o/m0AVsN1r4m1ZnZZWjo=
+decred.org/cspp/v2 v2.3.0/go.mod h1:9nO3bfvCheOPIFZw5f6sRQ42CjBFB5RKSaJ9Iq6G4MA=
 github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412 h1:w1UutsfOrms1J05zt7ISrnJIXKzwaspym5BTKGx93EI=
 github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412/go.mod h1:WPjqKcmVOxf0XSf3YxCJs6N6AOSrOx3obionmG7T0y0=
 github.com/companyzero/sntrup4591761 v0.0.0-20220309191932-9e0f3af2f07a h1:clYxJ3Os0EQUKDDVU8M0oipllX0EkuFNBfhVQuIfyF0=

mixing/mixclient/client.go

@@ -16,6 +16,7 @@ import (
 	"runtime"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"decred.org/cspp/v2/solverrpc"
@@ -36,7 +37,7 @@ import (
 // MinPeers is the minimum number of peers required for a mix run to proceed.
 const MinPeers = 4
 
-const pairingFlags byte = 0
+const pairingVersion byte = 1
 
 const (
 	timeoutDuration = 30 * time.Second
@@ -264,6 +265,9 @@ type queueWork struct {
 
 // Client manages local mixing client sessions.
 type Client struct {
+	// atomics
+	atomicPRFlags uint32
+
 	wallet  Wallet
 	mixpool *mixpool.Pool
 
@@ -281,8 +285,7 @@ type Client struct {
 	blake256Hasher   hash.Hash
 	blake256HasherMu sync.Mutex
 
-	epoch   time.Duration
-	prFlags byte
+	epoch time.Duration
 
 	logger slog.Logger
 
@@ -300,14 +303,14 @@ func NewClient(w Wallet) *Client {
 
 	height, _ := w.BestBlock()
 	return &Client{
+		atomicPRFlags:  uint32(prFlags),
 		wallet:         w,
 		mixpool:        w.Mixpool(),
 		pairings:       make(map[string]*pairedSessions),
 		warming:        make(chan struct{}),
 		workQueue:      make(chan *queueWork, runtime.NumCPU()),
 		blake256Hasher: blake256.New(),
 		epoch:          w.Mixpool().Epoch(),
-		prFlags:        prFlags,
 		height:         height,
 	}
 }
@@ -516,11 +519,11 @@ func (c *Client) prDelay(ctx context.Context, p *peer) error {
 	sendBefore := epoch.Add(-timeoutDuration - maxJitter)
 	sendAfter := epoch.Add(timeoutDuration)
 	var wait time.Duration
-	if now.After(sendBefore) {
+	if !now.Before(sendBefore) {
 		wait = sendAfter.Sub(now)
 		sendBefore = sendBefore.Add(c.epoch)
 	}
-	wait += p.msgJitter() + rand.Duration(time.Until(sendBefore))
+	wait += p.msgJitter() + rand.Duration(sendBefore.Sub(now))
 	timer := time.NewTimer(wait)
 	select {
 	case <-ctx.Done():
@@ -718,10 +721,11 @@ func (c *Client) Dicemix(ctx context.Context, cj *CoinJoin) error {
 		return err
 	}
 
+	prFlags := byte(atomic.LoadUint32(&c.atomicPRFlags))
 	pr, err := wire.NewMsgMixPairReq(*p.id, cj.prExpiry, cj.mixValue,
 		string(mixing.ScriptClassP2PKHv0), cj.tx.Version,
 		cj.tx.LockTime, cj.mcount, cj.inputValue, cj.prUTXOs,
-		cj.change, c.prFlags, pairingFlags)
+		cj.change, prFlags, pairingVersion)
 	if err != nil {
 		return err
 	}
@@ -1694,7 +1698,8 @@ func (c *Client) run(ctx context.Context, ps *pairedSessions, madePairing *bool)
 }
 
 func (c *Client) solvesRoots() bool {
-	return c.prFlags&mixing.PRFlagCanSolveRoots != 0
+	prFlags := byte(atomic.LoadUint32(&c.atomicPRFlags))
+	return prFlags&mixing.PRFlagCanSolveRoots != 0
 }
 
 // roots returns the solutions to the slot reservation polynomial.  If the
@@ -1704,8 +1709,20 @@ func (c *Client) solvesRoots() bool {
 func (c *Client) roots(ctx context.Context, seenSRs []chainhash.Hash,
 	sesRun *sessionRun, a []*big.Int, F *big.Int, publishedRoots chan struct{}) ([]*big.Int, error) {
 
-	if c.solvesRoots() {
+	switch {
+	case c.solvesRoots():
 		roots, err := solverrpc.Roots(a, F)
+		if errors.Is(err, solverrpc.ErrSolverProcessExited) {
+			// Unset root-solving capability and future advertisement
+			prFlags := byte(atomic.LoadUint32(&c.atomicPRFlags))
+			prFlags &^= mixing.PRFlagCanSolveRoots
+			atomic.StoreUint32(&c.atomicPRFlags, uint32(prFlags))
+
+			// Wait for other peers to publish roots as a fallback
+			// in this run.  If we are the only solver, we will be
+			// blamed for not publishing any.
+			break
+		}
 		if err != nil || len(roots) != len(a)-1 {
 			c.forLocalPeers(ctx, sesRun, func(p *peer) error {
 				p.triggeredBlame = true
@@ -1714,6 +1731,9 @@ func (c *Client) roots(ctx context.Context, seenSRs []chainhash.Hash,
 			close(publishedRoots)
 			return nil, errTriggeredBlame
 		}
+		sort.Slice(roots, func(i, j int) bool {
+			return roots[i].Cmp(roots[j]) == -1
+		})
 		rootBytes := make([][]byte, len(roots))
 		for i, root := range roots {
 			rootBytes[i] = root.Bytes()
@@ -1785,6 +1805,12 @@ func (c *Client) roots(ctx context.Context, seenSRs []chainhash.Hash,
 					roots = append(roots, root)
 				}
 			}
+			sorted := sort.SliceIsSorted(roots, func(i, j int) bool {
+				return roots[i].Cmp(roots[j]) == -1
+			})
+			if !sorted {
+				continue
+			}
 			if len(roots) == len(a)-1 {
 				return roots, nil
 			}

mixing/mixpool/mixpool.go

@@ -731,10 +731,13 @@ type Received struct {
 // such session has any messages currently accepted in the mixpool, the method
 // immediately errors.
 //
-// If any secrets messages are received for the described session, and r.RSs
-// is nil, Receive immediately returns ErrSecretsRevealed.  An additional call
-// to Receive with a non-nil RSs can be used to receive all of the secrets
-// after each peer publishes their own revealed secrets.
+// If r.RSs is nil and any secrets messages are received for the described
+// session, Receive will count these messages towards the total number of
+// expected messages based on the slice capacities (without overcounting if
+// the identity has also sent an expected non-RS message) and will return
+// ErrSecretsRevealed.  An additional call to Receive with a non-nil RSs can
+// be used to receive all of the secrets after each peer publishes their own
+// revealed secrets.
 func (p *Pool) Receive(ctx context.Context, r *Received) error {
 	sid := r.Sid
 	var bc *broadcast
@@ -784,42 +787,32 @@ Loop:
 	for {
 		// Pool is locked for reads.  Count if the total number of
 		// expected messages have been received.
-		received := 0
+		received := make(map[idPubKey]struct{})
+		countMsg := func(msg mixing.Message) {
+			received[*(*idPubKey)(msg.Pub())] = struct{}{}
+		}
 		for hash := range ses.hashes {
-			msgtype := p.pool[hash].msgtype
+			e := p.pool[hash]
+			msg := e.msg
+			msgtype := e.msgtype
 			switch {
 			case msgtype == msgtypeKE && r.KEs != nil:
-				received++
+				countMsg(msg)
 			case msgtype == msgtypeCT && r.CTs != nil:
-				received++
+				countMsg(msg)
 			case msgtype == msgtypeSR && r.SRs != nil:
-				received++
+				countMsg(msg)
 			case msgtype == msgtypeDC && r.DCs != nil:
-				received++
+				countMsg(msg)
 			case msgtype == msgtypeCM && r.CMs != nil:
-				received++
+				countMsg(msg)
 			case msgtype == msgtypeFP && r.FPs != nil:
-				received++
+				countMsg(msg)
 			case msgtype == msgtypeRS:
-				if r.RSs == nil {
-					// Since initial reporters of secrets
-					// need to take the blame for
-					// erroneous blame assignment if no
-					// issue was detected, we only trigger
-					// this for RS messages that do not
-					// reference any other previous RS.
-					rs := p.pool[hash].msg.(*wire.MsgMixSecrets)
-					prev := rs.PrevMsgs()
-					if len(prev) == 0 {
-						p.mtx.RUnlock()
-						return ErrSecretsRevealed
-					}
-				} else {
-					received++
-				}
+				countMsg(msg)
 			}
 		}
-		if received >= expectedMessages {
+		if len(received) >= expectedMessages {
 			break
 		}
 
@@ -836,6 +829,8 @@ Loop:
 		p.mtx.RLock()
 	}
 
+	var err error
+
 	// Pool is locked for reads.  Collect all of the messages.
 	for hash := range ses.hashes {
 		msg := p.pool[hash].msg
@@ -865,14 +860,16 @@ Loop:
 				r.FPs = append(r.FPs, msg)
 			}
 		case *wire.MsgMixSecrets:
-			if r.RSs != nil {
+			if r.RSs == nil {
+				err = ErrSecretsRevealed
+			} else {
 				r.RSs = append(r.RSs, msg)
 			}
 		}
 	}
 
 	p.mtx.RUnlock()
-	return nil
+	return err
 }
 
 var zeroHash chainhash.Hash