chain: Fetch unknown PRs of notified orphan KEs

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.")
@@ -947,5 +935,28 @@ func (s *Syncer) mixMessage(ctx context.Context, params json.RawMessage) error {
 		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) {