chain: Fetch unknown PRs of notified orphan KEs

This changes the notification handler when a run-0 KE is received but the
sender's own PR is not yet known to request the PR from dcrd.

Because this also works to initially sync the mixpool during the warming
period, the previous call to getmixpairrequests is no longer necessary and has
been removed.

chain/sync.go

@@ -25,12 +25,13 @@ import (
 	"github.com/decred/dcrd/blockchain/stake/v5"
 	"github.com/decred/dcrd/chaincfg/chainhash"
 	"github.com/decred/dcrd/crypto/blake256"
+	"github.com/decred/dcrd/mixing/mixpool"
 	"github.com/decred/dcrd/wire"
 	"github.com/jrick/wsrpc/v2"
 	"golang.org/x/sync/errgroup"
 )
 
-var requiredAPIVersion = semver{Major: 8, Minor: 2, Patch: 0}
+var requiredAPIVersion = semver{Major: 8, Minor: 3, Patch: 0}
 
 // Syncer implements wallet synchronization services by processing
 // notifications from a dcrd JSON-RPC server.
@@ -640,11 +641,6 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 			err = e
 		}
 	}()
-	g.Go(func() error {
-		// Run wallet background goroutines (currently, this just runs
-		// mixclient).
-		return s.wallet.Run(ctx)
-	})
 
 	if s.wallet.VotingEnabled() {
 		err = s.rpc.Call(ctx, "notifywinningtickets", nil)
@@ -670,6 +666,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 	s.fetchMissingCfiltersFinished()
 
 	// Fetch all headers the wallet has not processed yet.
+	// XXX: this also loads active addresses.
 	err = s.getHeaders(ctx)
 	if err != nil {
 		return err
@@ -713,27 +710,18 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		return err
 	}
 
+	g.Go(func() error {
+		// Run wallet background goroutines (currently, this just runs
+		// mixclient).
+		return s.wallet.Run(ctx)
+	})
+
 	// Request notifications for mixing messages.
 	if s.wallet.MixingEnabled() {
 		err = s.rpc.Call(ctx, "notifymixmessages", nil)
 		if err != nil {
 			return err
 		}
-
-		// Populate existing mix pair requests.
-		mixPRs, err := s.rpc.MixPairRequests(ctx)
-		if err != nil {
-			return err
-		}
-		s.blake256HasherMu.Lock()
-		for _, pr := range mixPRs {
-			pr.WriteHash(s.blake256Hasher)
-		}
-		s.blake256HasherMu.Unlock()
-		for _, pr := range mixPRs {
-			loggers.MixcLog.Debugf("accepting PR hash %s at initial sync", pr.Hash())
-			s.wallet.AcceptMixMessage(pr)
-		}
 	}
 
 	log.Infof("Blockchain sync completed, wallet ready for general usage.")
@@ -946,6 +934,30 @@ func (s *Syncer) mixMessage(ctx context.Context, params json.RawMessage) error {
 	} else {
 		loggers.MixpLog.Debugf("Rejected mix message %T %s by %x",
 			msg, &msgHash, msg.Pub())
+
 	}
+
+	var e *mixpool.MissingOwnPRError
+	if errors.As(err, &e) {
+		ke, ok := msg.(*wire.MsgMixKeyExchange)
+		if !ok || ke.Run != 0 {
+			return err
+		}
+		pr, err := s.rpc.MixMessage(ctx, &e.MissingPR)
+		if err == nil {
+			s.blake256HasherMu.Lock()
+			pr.WriteHash(s.blake256Hasher)
+			s.blake256HasherMu.Unlock()
+			prHash := pr.Hash()
+
+			err = s.wallet.AcceptMixMessage(pr)
+			if err == nil {
+				loggers.MixpLog.Debugf("Accepted missing PR %s for "+
+					"previous orphan KE %s", &prHash, &msgHash)
+			}
+		}
+		return err
+	}
+
 	return err
 }

go.mod

@@ -20,8 +20,8 @@ require (
 	github.com/decred/dcrd/dcrutil/v4 v4.0.2
 	github.com/decred/dcrd/gcs/v4 v4.1.0
 	github.com/decred/dcrd/hdkeychain/v3 v3.1.2
-	github.com/decred/dcrd/mixing v0.1.1-0.20240518194732-ac51ffa827c1
-	github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.2.0
+	github.com/decred/dcrd/mixing v0.1.1-0.20240524012907-0952040fe22d
+	github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.2.1-0.20240524012907-0952040fe22d
 	github.com/decred/dcrd/rpcclient/v8 v8.0.1
 	github.com/decred/dcrd/txscript/v4 v4.1.1
 	github.com/decred/dcrd/wire v1.7.0

go.sum

@@ -59,12 +59,10 @@ github.com/decred/dcrd/gcs/v4 v4.1.0 h1:tpW7JW53yJZlgNwl/n2NL1b8NxHaIPRUyNuLMkB/
 github.com/decred/dcrd/gcs/v4 v4.1.0/go.mod h1:nPTbGM/I3Ihe5KFvUmxZEqQP/jDZQjQ63+WEi/f4lqU=
 github.com/decred/dcrd/hdkeychain/v3 v3.1.2 h1:x25WuuE7zM/20EynuVMyOhL0K8BwGBBsexGq8xTiHFA=
 github.com/decred/dcrd/hdkeychain/v3 v3.1.2/go.mod h1:FnNJmZ7jqUDeAo6/c/xkQi5cuxh3EWtJeMmW6/Z8lcc=
-github.com/decred/dcrd/mixing v0.1.0 h1:XmRdBipQE7dSpY/5VAvI2EOqGu9bzAh2KzS0JSZ1CIs=
-github.com/decred/dcrd/mixing v0.1.0/go.mod h1:W3K7yJKmoI03G2U5Yw+HSRNe6lLBegi63ZR6fFLnM9c=
-github.com/decred/dcrd/mixing v0.1.1-0.20240518194732-ac51ffa827c1 h1:4mzKsOnVI3Ugq5KX1fGB5+nKUoNmpLnhWVVciOYGR0s=
-github.com/decred/dcrd/mixing v0.1.1-0.20240518194732-ac51ffa827c1/go.mod h1:W3K7yJKmoI03G2U5Yw+HSRNe6lLBegi63ZR6fFLnM9c=
-github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.2.0 h1:BmxdaVwddO5UeYkWFO1JIZiQe8kE1DdNkohlLx9aDec=
-github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.2.0/go.mod h1:dDHO7ivrPAhZjFD3LoOJN/kdq5gi0sxie6zCsWHAiUo=
+github.com/decred/dcrd/mixing v0.1.1-0.20240524012907-0952040fe22d h1:9zBmaveBE5OIH617PxJg073F3S/j58sxrc8rXYj3Yhw=
+github.com/decred/dcrd/mixing v0.1.1-0.20240524012907-0952040fe22d/go.mod h1:W3K7yJKmoI03G2U5Yw+HSRNe6lLBegi63ZR6fFLnM9c=
+github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.2.1-0.20240524012907-0952040fe22d h1:6bOsyNqgxcaRVhzimPlR/K8wWSuB07OwVvmzj6AYOkk=
+github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.2.1-0.20240524012907-0952040fe22d/go.mod h1:dDHO7ivrPAhZjFD3LoOJN/kdq5gi0sxie6zCsWHAiUo=
 github.com/decred/dcrd/rpcclient/v8 v8.0.1 h1:hd81e4w1KSqvPcozJlnz6XJfWKDNuahgooH/N5E8vOU=
 github.com/decred/dcrd/rpcclient/v8 v8.0.1/go.mod h1:97XD5P/XrZzedePPFPJzc8el2o00q2Kr+Epi4AvRL3o=
 github.com/decred/dcrd/txscript/v4 v4.1.1 h1:R4M2+jMujgQA91899SkL0cW66d6DC76Gx+1W1oEHjc0=

rpc/client/dcrd/calls.go

@@ -249,6 +249,43 @@ func (r *RPC) PublishMixMessages(ctx context.Context, msgs ...mixing.Message) er
 	return nil
 }
 
+// MixMessage queries the dcrd mixpool for a mixing message by its hash.
+func (r *RPC) MixMessage(ctx context.Context, hash *chainhash.Hash) (mixing.Message, error) {
+	const op errors.Op = "dcrd.MixMessage"
+
+	var mixMessage *dcrdtypes.GetMixMessageResult
+	err := r.Call(ctx, "getmixmessage", &mixMessage, hash.String())
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	var msg mixing.Message
+	switch mixMessage.Type {
+	case wire.CmdMixPairReq:
+		msg = new(wire.MsgMixPairReq)
+	case wire.CmdMixKeyExchange:
+		msg = new(wire.MsgMixKeyExchange)
+	case wire.CmdMixCiphertexts:
+		msg = new(wire.MsgMixCiphertexts)
+	case wire.CmdMixSlotReserve:
+		msg = new(wire.MsgMixSlotReserve)
+	case wire.CmdMixDCNet:
+		msg = new(wire.MsgMixDCNet)
+	case wire.CmdMixConfirm:
+		msg = new(wire.MsgMixConfirm)
+	case wire.CmdMixFactoredPoly:
+		msg = new(wire.MsgMixFactoredPoly)
+	case wire.CmdMixSecrets:
+		msg = new(wire.MsgMixSecrets)
+	default:
+		err = errors.E(op, "unrecognized mixing message type")
+		return nil, err
+	}
+
+	err = msg.BtcDecode(hex.NewDecoder(strings.NewReader(mixMessage.Message)), wire.MixVersion)
+	return msg, err
+}
+
 // MixPairRequests returns all mixing pair request messages currently held by
 // the dcrd mixpool.
 func (r *RPC) MixPairRequests(ctx context.Context) ([]*wire.MsgMixPairReq, error) {