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

@@ -58,3 +58,7 @@ require (
 	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
 	lukechampine.com/blake3 v1.3.0 // indirect
 )
+
+replace github.com/decred/dcrd/mixing => github.com/jrick/dcrd/mixing v0.0.0-20240523192744-ba3a9c2209c5
+
+replace github.com/decred/dcrd/rpc/jsonrpc/types/v4 => github.com/jrick/dcrd/rpc/jsonrpc/types/v4 v4.0.0-20240523192744-ba3a9c2209c5

go.sum

@@ -59,12 +59,6 @@ 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/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=
@@ -118,6 +112,10 @@ github.com/jessevdk/go-flags v1.5.0 h1:1jKYvbxEjfUl0fmqTCOfonvskHHXMjBySTLW4y9LF
 github.com/jessevdk/go-flags v1.5.0/go.mod h1:Fw0T6WPc1dYxT4mKEZRfG5kJhaTDP9pj1c2EWnYs/m4=
 github.com/jrick/bitset v1.0.0 h1:Ws0PXV3PwXqWK2n7Vz6idCdrV/9OrBXgHEJi27ZB9Dw=
 github.com/jrick/bitset v1.0.0/go.mod h1:ZOYB5Uvkla7wIEY4FEssPVi3IQXa02arznRaYaAEPe4=
+github.com/jrick/dcrd/mixing v0.0.0-20240523192744-ba3a9c2209c5 h1:wUnGO5J75bLYtVCZbx0cntzMoFuEZwBhqu3F8Puf6Xs=
+github.com/jrick/dcrd/mixing v0.0.0-20240523192744-ba3a9c2209c5/go.mod h1:W3K7yJKmoI03G2U5Yw+HSRNe6lLBegi63ZR6fFLnM9c=
+github.com/jrick/dcrd/rpc/jsonrpc/types/v4 v4.0.0-20240523192744-ba3a9c2209c5 h1:q8JbhTvS25TYcwqY9dEOvmPaQnqkqNL+b4gedDBzjJ4=
+github.com/jrick/dcrd/rpc/jsonrpc/types/v4 v4.0.0-20240523192744-ba3a9c2209c5/go.mod h1:dDHO7ivrPAhZjFD3LoOJN/kdq5gi0sxie6zCsWHAiUo=
 github.com/jrick/logrotate v1.0.0 h1:lQ1bL/n9mBNeIXoTUoYRlK4dHuNJVofX9oWqBtPnSzI=
 github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
 github.com/jrick/wsrpc/v2 v2.3.5 h1:CwdycaR/df09iGkPMXs1FxqAHMCQbdAiTGoHfOrtuds=

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) {