Order mixclient and syncer shutdown

chain/sync.go

@@ -542,9 +542,11 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 
 	params := s.wallet.ChainParams()
 
+	ntfnCtx, ntfnCtxCancel := context.WithCancel(context.Background())
+	defer ntfnCtxCancel()
 	s.notifier = &notifier{
 		syncer: s,
-		ctx:    ctx,
+		ctx:    ntfnCtx,
 		closed: make(chan struct{}),
 	}
 	addr, err := normalizeAddress(s.opts.Address, s.opts.DefaultPort)
@@ -589,12 +591,12 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		}
 		opts = append(opts, wsrpc.WithTLSConfig(tc))
 	}
-	client, err := wsrpc.Dial(ctx, addr, opts...)
+	wsClient, err := wsrpc.Dial(ctx, addr, opts...)
 	if err != nil {
 		return err
 	}
-	defer client.Close()
-	s.rpc = dcrd.New(client)
+	defer wsClient.Close()
+	s.rpc = dcrd.New(wsClient)
 
 	// Verify that the server is running on the expected network.
 	var netID wire.CurrencyNet
@@ -723,10 +725,27 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		return err
 	}
 
+	defer func() {
+		ntfnCtxCancel()
+
+		select {
+		case <-ctx.Done():
+			wsClient.Close()
+		default:
+		}
+
+		// Wait for notifications to finish before returning
+		<-s.notifier.closed
+	}()
+
+	// Ensure wallet.Run cleanly finishes/is canceled first when outer
+	// context is canceled.
+	walletCtx, walletCtxCancel := context.WithCancel(context.Background())
+	defer walletCtxCancel()
 	g.Go(func() error {
 		// Run wallet background goroutines (currently, this just runs
 		// mixclient).
-		return s.wallet.Run(ctx)
+		return s.wallet.Run(walletCtx)
 	})
 
 	// Request notifications for mixing messages.
@@ -739,18 +758,13 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 
 	log.Infof("Blockchain sync completed, wallet ready for general usage.")
 
-	// Wait for notifications to finish before returning
-	defer func() {
-		<-s.notifier.closed
-	}()
-
 	g.Go(func() error {
 		select {
 		case <-ctx.Done():
-			client.Close()
+			walletCtxCancel()
 			return ctx.Err()
-		case <-client.Done():
-			return client.Err()
+		case <-wsClient.Done():
+			return wsClient.Err()
 		}
 	})
 	return g.Wait()

dcrwallet.go

@@ -534,9 +534,10 @@ func spvLoop(ctx context.Context, w *wallet.Wallet) {
 	for {
 		err := syncer.Run(ctx)
 		if done(ctx) {
+			loggers.SyncLog.Infof("SPV synchronization stopped")
 			return
 		}
-		log.Errorf("SPV synchronization ended: %v", err)
+		loggers.SyncLog.Errorf("SPV synchronization stopped: %v", err)
 	}
 }
 
@@ -571,7 +572,11 @@ func rpcSyncLoop(ctx context.Context, w *wallet.Wallet) {
 		syncer := chain.NewSyncer(w, rpcOptions)
 		err := syncer.Run(ctx)
 		if err != nil {
-			loggers.SyncLog.Errorf("Wallet synchronization stopped: %v", err)
+			if errors.Is(err, context.Canceled) || ctx.Err() != nil {
+				loggers.SyncLog.Infof("RPC synchronization stopped")
+				return
+			}
+			loggers.SyncLog.Errorf("RPC synchronization stopped: %v", err)
 			select {
 			case <-ctx.Done():
 				return

go.mod

@@ -59,3 +59,5 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241223144023-3abc09e42ca8 // indirect
 	lukechampine.com/blake3 v1.3.0 // indirect
 )
+
+replace github.com/decred/dcrd/mixing => github.com/jrick/dcrd/mixing v0.0.0-20250123211715-8a5ce5c2063a

go.sum

@@ -50,8 +50,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.4.2 h1:mpt2pNIFTI6L1hXrieAWJTQJv5t9WzHcNnhI+tnAG90=
-github.com/decred/dcrd/mixing v0.4.2/go.mod h1:VF87lOn41kitgWVOwmXoB4qMYF7+bxItZXyw4JfW3EQ=
 github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.3.0 h1:l0DnCcILTNrpy8APF3FLN312ChpkQaAuW30aC/RgBaw=
 github.com/decred/dcrd/rpc/jsonrpc/types/v4 v4.3.0/go.mod h1:j+kkRPXPJB5S9VFOsx8SQLcU7PTFkPKRc1aCHN4ENzA=
 github.com/decred/dcrd/rpcclient/v8 v8.0.1 h1:hd81e4w1KSqvPcozJlnz6XJfWKDNuahgooH/N5E8vOU=
@@ -78,6 +76,8 @@ 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-20250123211715-8a5ce5c2063a h1:wGTjDa+kmjKBkhnc8BACcU+OreLC+gxTFHLz5t+AFkw=
+github.com/jrick/dcrd/mixing v0.0.0-20250123211715-8a5ce5c2063a/go.mod h1:VF87lOn41kitgWVOwmXoB4qMYF7+bxItZXyw4JfW3EQ=
 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.8 h1:9vfM8o9g00HXQb/3D6+Y9Cy1uybjD7K1272vtdXXBps=

spv/sync.go

@@ -321,7 +321,7 @@ func (s *Syncer) setRequiredHeight(tipHeight int32) {
 }
 
 // Run synchronizes the wallet, returning when synchronization fails or the
-// context is cancelled.
+// context is canceled.
 func (s *Syncer) Run(ctx context.Context) (err error) {
 	s.doneMu.Lock()
 	s.done = make(chan struct{})
@@ -367,23 +367,23 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 	}
 
 	// Start background handlers to read received messages from remote peers
-	g, ctx := errgroup.WithContext(ctx)
-	g.Go(func() error { return s.receiveGetData(ctx) })
-	g.Go(func() error { return s.receiveInv(ctx) })
-	g.Go(func() error { return s.receiveHeadersAnnouncements(ctx) })
-	g.Go(func() error { return s.receiveMixMsgs(ctx) })
+	g, gctx := errgroup.WithContext(context.Background())
+	g.Go(func() error { return s.receiveGetData(gctx) })
+	g.Go(func() error { return s.receiveInv(gctx) })
+	g.Go(func() error { return s.receiveHeadersAnnouncements(gctx) })
+	g.Go(func() error { return s.receiveMixMsgs(gctx) })
 	s.lp.AddHandledMessages(p2p.MaskGetData | p2p.MaskInv)
 
 	if len(s.persistentPeers) != 0 {
 		for i := range s.persistentPeers {
 			raddr := s.persistentPeers[i]
-			g.Go(func() error { return s.connectToPersistent(ctx, raddr) })
+			g.Go(func() error { return s.connectToPersistent(gctx, raddr) })
 		}
 	} else {
-		g.Go(func() error { return s.connectToCandidates(ctx) })
+		g.Go(func() error { return s.connectToCandidates(gctx) })
 	}
 
-	g.Go(func() error { return s.handleMempool(ctx) })
+	g.Go(func() error { return s.handleMempool(gctx) })
 
 	s.wallet.SetNetworkBackend(s)
 	defer s.wallet.SetNetworkBackend(nil)
@@ -392,7 +392,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 	g.Go(func() error {
 		// First step: fetch missing CFilters.
 		progress := make(chan wallet.MissingCFilterProgress, 1)
-		go s.wallet.FetchMissingCFiltersWithProgress(ctx, s, progress)
+		go s.wallet.FetchMissingCFiltersWithProgress(gctx, s, progress)
 
 		log.Debugf("Fetching missing CFilters...")
 		s.fetchMissingCfiltersStart()
@@ -408,14 +408,14 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		// Next: fetch headers and cfilters up to mainchain tip.
 		s.fetchHeadersStart()
 		log.Debugf("Fetching headers and CFilters...")
-		err = s.initialSyncHeaders(ctx)
+		err = s.initialSyncHeaders(gctx)
 		if err != nil {
 			return err
 		}
 		s.fetchHeadersFinished()
 
 		// Finally: Perform the initial rescan over the received blocks.
-		err = s.initialSyncRescan(ctx)
+		err = s.initialSyncRescan(gctx)
 		if err != nil {
 			return err
 		}
@@ -425,10 +425,22 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		return nil
 	})
 
-	// Run wallet background goroutines (currently, this just runs
-	// mixclient).
+	// Ensure wallet.Run cleanly finishes/is canceled first when outer
+	// context is canceled.
+	walletCtx, walletCtxCancel := context.WithCancel(context.Background())
+	g.Go(func() error {
+		<-ctx.Done()
+		walletCtxCancel()
+		return nil // Do not cancel gctx yet.
+	})
 	g.Go(func() error {
-		return s.wallet.Run(ctx)
+		// Run wallet background goroutines (currently, this just runs
+		// mixclient).
+		err := s.wallet.Run(walletCtx)
+		if ctxErr := ctx.Err(); err == nil && ctxErr != nil {
+			return ctxErr // Cancels gctx, if outer ctx has errored/canceled.
+		}
+		return err
 	})
 
 	// Wait until cancellation or a handler errors.