func (c *Client) rollWork() {
	ticker := time.NewTicker(c.cfg.RollWorkCycle)
	defer ticker.Stop()
	for {
		select {
		case <-c.ctx.Done():
			c.wg.Done()
			return
		case <-ticker.C:
			lastWorkTime := atomic.LoadInt64(&c.lastWorkTime)
			if lastWorkTime == 0 {
				continue
			}
			now := time.Now()
			if now.Sub(time.Unix(lastWorkTime, 0)) >= c.cfg.MaxGenTime*2 {
				c.updateWork(false)
			}
		}
	}
}
rollWork provides the client with timestamp-rolled work to avoid stalling.
package pool
import (
	"bufio"
	"bytes"
	"context"
	"crypto/rand"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math/big"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/decred/dcrd/blockchain/standalone/v2"
	"github.com/decred/dcrd/chaincfg/chainhash"
	"github.com/decred/dcrd/chaincfg/v3"
	"github.com/decred/dcrd/dcrutil/v3"
	"github.com/decred/dcrd/wire"
	errs "github.com/decred/dcrpool/errors"
)
const (
	maxMessageSize = 250
	hashCalcThreshold = time.Second * 20
	rollWorkCycle = time.Second
)
var (
	ZeroInt = new(big.Int).SetInt64(0)
	ZeroRat = new(big.Rat).SetInt64(0)
	zeroHash = chainhash.Hash{0}
)
type readPayload struct {
	msg     Message
	msgType int
}
type ClientConfig struct {
	ActiveNet *chaincfg.Params
	db Database
	SoloPool bool
	Blake256Pad []byte
	NonceIterations float64
	FetchMinerDifficulty func(string) (*DifficultyInfo, error)
	Disconnect func()
	RemoveClient func(*Client)
	SubmitWork func(context.Context, *string) (bool, error)
	FetchCurrentWork func() string
	WithinLimit func(string, int) bool
	HashCalcThreshold time.Duration
	MaxGenTime time.Duration
	ClientTimeout time.Duration
	SignalCache func(event CacheUpdateEvent)
	MonitorCycle time.Duration
	MaxUpgradeTries uint32
	RollWorkCycle time.Duration
}
type Client struct {
	submissions  int64  
	lastWorkTime int64  
	miner    string
	id       string
	diffInfo *DifficultyInfo
	mtx      sync.RWMutex
	addr        *net.TCPAddr
	cfg         *ClientConfig
	conn        net.Conn
	encoder     *json.Encoder
	reader      *bufio.Reader
	ctx         context.Context
	cancel      context.CancelFunc
	name        string
	extraNonce1 string
	ch          chan Message
	readCh      chan readPayload
	account     string
	authorized bool
	subscribed bool
	statusMtx  sync.RWMutex
	hashRate    *big.Rat
	hashRateMtx sync.RWMutex
	wg          sync.WaitGroup
}
func (c *Client) generateExtraNonce1() error {
	id := make([]byte, 4)
	_, err := rand.Read(id)
	if err != nil {
		return err
	}
	c.extraNonce1 = hex.EncodeToString(id)
	return nil
}
func NewClient(ctx context.Context, conn net.Conn, addr *net.TCPAddr, cCfg *ClientConfig) (*Client, error) {
	ctx, cancel := context.WithCancel(ctx)
	c := &Client{
		addr:     addr,
		cfg:      cCfg,
		conn:     conn,
		ctx:      ctx,
		cancel:   cancel,
		ch:       make(chan Message),
		readCh:   make(chan readPayload),
		encoder:  json.NewEncoder(conn),
		reader:   bufio.NewReaderSize(conn, maxMessageSize),
		hashRate: ZeroRat,
	}
	err := c.generateExtraNonce1()
	if err != nil {
		return nil, err
	}
	return c, nil
}
func (c *Client) shutdown() {
	c.cfg.RemoveClient(c)
	c.mtx.RLock()
	id := c.id
	c.mtx.RUnlock()
	log.Tracef("%s connection terminated.", id)
}
func (c *Client) claimWeightedShare() error {
	if c.cfg.SoloPool {
		desc := "cannot claim shares in solo pool mode"
		return errs.PoolError(errs.ClaimShare, desc)
	}
	c.mtx.RLock()
	miner := c.miner
	c.mtx.RUnlock()
	if c.cfg.ActiveNet.Name == chaincfg.MainNetParams().Name && miner == CPU {
		desc := "cannot claim shares for cpu miners on mainnet, " +
			"reserved for testing purposes only (simnet, testnet)"
		return errs.PoolError(errs.ClaimShare, desc)
	}
	weight := ShareWeights[miner]
	share := NewShare(c.account, weight)
	return c.cfg.db.PersistShare(share)
}
func (c *Client) handleAuthorizeRequest(req *Request, allowed bool) error {
	if !allowed {
		err := fmt.Errorf("unable to process authorize request, " +
			"client request limit reached")
		sErr := NewStratumError(Unknown, err)
		resp := AuthorizeResponse(*req.ID, false, sErr)
		c.ch <- resp
		return errs.PoolError(errs.LimitExceeded, err.Error())
	}
	username, err := ParseAuthorizeRequest(req)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := AuthorizeResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	switch c.cfg.SoloPool {
	case false:
		parts := strings.Split(username, ".")
		if len(parts) != 2 {
			err := fmt.Errorf("invalid username format, expected "+
				"`address.clientid`, got %v", username)
			sErr := NewStratumError(Unknown, err)
			resp := AuthorizeResponse(*req.ID, false, sErr)
			c.ch <- resp
			return errs.MsgError(errs.Parse, err.Error())
		}
		name := strings.TrimSpace(parts[1])
		address := strings.TrimSpace(parts[0])
		_, err = dcrutil.DecodeAddress(address, c.cfg.ActiveNet)
		if err != nil {
			sErr := NewStratumError(Unknown, err)
			resp := AuthorizeResponse(*req.ID, false, sErr)
			c.ch <- resp
			return err
		}
		account := NewAccount(address)
		err = c.cfg.db.persistAccount(account)
		if err != nil {
			if !errors.Is(err, errs.ValueFound) {
				sErr := NewStratumError(Unknown, err)
				resp := AuthorizeResponse(*req.ID, false, sErr)
				c.ch <- resp
				return err
			}
		}
		c.account = account.UUID
		c.name = name
	case true:
		c.account = defaultAccountID
		c.name = username
	}
	c.statusMtx.Lock()
	c.authorized = true
	c.statusMtx.Unlock()
	resp := AuthorizeResponse(*req.ID, true, nil)
	c.ch <- resp
	return nil
}
func (c *Client) monitor(idx int, pair *minerIDPair, monitorCycle time.Duration, maxTries uint32) {
	var subs, tries uint32
	if len(pair.miners) <= 1 {
		return
	}
	expected := float64(monitorCycle / c.cfg.MaxGenTime)
	for {
		ticker := time.NewTicker(monitorCycle)
		defer ticker.Stop()
		select {
		case <-ticker.C:
			if idx == len(pair.miners)-1 {
				return
			}
			if tries == maxTries {
				return
			}
			subs = uint32(atomic.LoadInt64(&c.submissions)) - subs
			delta := float64(subs) - expected
			if delta < 0.0 || delta < expected*0.3 {
				tries++
				continue
			}
			idx++
			c.mtx.Lock()
			miner := pair.miners[idx]
			newID := fmt.Sprintf("%v/%v", c.extraNonce1, miner)
			log.Infof("upgrading %s to %s", c.id, newID)
			info, err := c.cfg.FetchMinerDifficulty(miner)
			if err != nil {
				tries++
				log.Error(err)
				c.mtx.Unlock()
				continue
			}
			c.miner = miner
			c.id = newID
			c.diffInfo = info
			c.mtx.Unlock()
			c.setDifficulty()
			log.Infof("updated difficulty (%s) for %s sent",
				c.diffInfo.difficulty.FloatString(3), c.id)
			c.updateWork(true)
			tries = 0
		case <-c.ctx.Done():
			return
		}
	}
}
func (c *Client) handleSubscribeRequest(req *Request, allowed bool) error {
	if !allowed {
		err := fmt.Errorf("unable to process subscribe request, client " +
			"request limit reached")
		sErr := NewStratumError(Unknown, err)
		resp := SubscribeResponse(*req.ID, "", "", 0, sErr)
		c.ch <- resp
		return errs.PoolError(errs.LimitExceeded, err.Error())
	}
	mid, nid, err := ParseSubscribeRequest(req)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubscribeResponse(*req.ID, "", "", 0, sErr)
		c.ch <- resp
		return err
	}
	idPair, err := identifyMiner(mid)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubscribeResponse(*req.ID, "", "", 0, sErr)
		c.ch <- resp
		return errs.PoolError(errs.MinerUnknown, err.Error())
	}
	c.mtx.Lock()
	minerIdx := 0
	miner := idPair.miners[minerIdx]
	info, err := c.cfg.FetchMinerDifficulty(miner)
	if err != nil {
		c.mtx.Unlock()
		return err
	}
	c.miner = miner
	c.id = fmt.Sprintf("%v/%v", c.extraNonce1, miner)
	c.diffInfo = info
	c.mtx.Unlock()
	if nid == "" {
		nid = fmt.Sprintf("mn%v", c.extraNonce1)
	}
	go c.monitor(minerIdx, idPair, c.cfg.MonitorCycle, c.cfg.MaxUpgradeTries)
	var resp *Response
	switch miner {
	case ObeliskDCR1:
		resp = SubscribeResponse(*req.ID, nid, c.extraNonce1,
			ExtraNonce2Size, nil)
	case AntminerDR3, AntminerDR5:
		paddedExtraNonce1 := strings.Repeat("0", 16) + c.extraNonce1
		resp = SubscribeResponse(*req.ID, nid, paddedExtraNonce1, 8, nil)
	case WhatsminerD1:
		paddedExtraNonce1 := strings.Repeat("0", 8) + c.extraNonce1
		resp = SubscribeResponse(*req.ID, nid, paddedExtraNonce1,
			ExtraNonce2Size, nil)
	default:
		resp = SubscribeResponse(*req.ID, nid, c.extraNonce1, ExtraNonce2Size, nil)
	}
	c.statusMtx.Lock()
	c.subscribed = true
	c.statusMtx.Unlock()
	c.ch <- resp
	return nil
}
func (c *Client) handleExtraNonceSubscribeRequest(req *Request, allowed bool) error {
	if !allowed {
		err := fmt.Errorf("unable to process extranonce subscribe request," +
			"client request limit reached")
		sErr := NewStratumError(Unknown, err)
		resp := SubscribeResponse(*req.ID, "", "", 0, sErr)
		c.ch <- resp
		return errs.PoolError(errs.LimitExceeded, err.Error())
	}
	err := ParseExtraNonceSubscribeRequest(req)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubscribeResponse(*req.ID, "", "", 0, sErr)
		c.ch <- resp
		return err
	}
	resp := ExtraNonceSubscribeResponse(*req.ID)
	c.ch <- resp
	return nil
}
func (c *Client) setDifficulty() {
	if c.diffInfo == nil {
		return
	}
	c.mtx.RLock()
	diffRat := c.diffInfo.difficulty
	c.mtx.RUnlock()
	diff := new(big.Rat).Set(diffRat)
	diffNotif := SetDifficultyNotification(diff)
	c.ch <- diffNotif
}
func (c *Client) handleSubmitWorkRequest(ctx context.Context, req *Request, allowed bool) error {
	if !allowed {
		err := fmt.Errorf("unable to process submit work request, client " +
			"request limit reached")
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return errs.PoolError(errs.LimitExceeded, err.Error())
	}
	c.mtx.RLock()
	id := c.id
	miner := c.miner
	powLimit := c.diffInfo.powLimit
	diff := c.diffInfo.difficulty
	tgt := c.diffInfo.target
	c.mtx.RUnlock()
	_, jobID, extraNonce2E, nTimeE, nonceE, err :=
		ParseSubmitWorkRequest(req, miner)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	job, err := c.cfg.db.fetchJob(jobID)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	header, err := GenerateSolvedBlockHeader(job.Header, c.extraNonce1,
		extraNonce2E, nTimeE, nonceE, miner)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	target := new(big.Rat).SetInt(standalone.CompactToBig(header.Bits))
	if target.Sign() <= 0 {
		err := fmt.Errorf("block target difficulty of %064x is too "+
			"low", target)
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return errs.PoolError(errs.LowDifficulty, err.Error())
	}
	hash := header.BlockHash()
	hashTarget := new(big.Rat).SetInt(standalone.HashToBig(&hash))
	netDiff := new(big.Rat).Quo(powLimit, target)
	hashDiff := new(big.Rat).Quo(powLimit, hashTarget)
	log.Tracef("network difficulty is: %s", netDiff.FloatString(4))
	log.Tracef("pool difficulty is: %s", diff.FloatString(4))
	log.Tracef("hash difficulty is: %s", hashDiff.FloatString(4))
	if hashTarget.Cmp(tgt) > 0 {
		err := fmt.Errorf("submitted work %s from %s is not less than its "+
			"corresponding pool target", hash.String(), id)
		sErr := NewStratumError(LowDifficultyShare, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return errs.PoolError(errs.PoolDifficulty, err.Error())
	}
	atomic.AddInt64(&c.submissions, 1)
	if !c.cfg.SoloPool {
		err := c.claimWeightedShare()
		if err != nil {
			err := fmt.Errorf("%s: %v", id, err)
			sErr := NewStratumError(Unknown, err)
			resp := SubmitWorkResponse(*req.ID, false, sErr)
			c.ch <- resp
			return errs.PoolError(errs.ClaimShare, err.Error())
		}
		c.cfg.SignalCache(ClaimedShare)
	}
	if hashTarget.Cmp(target) > 0 {
		resp := SubmitWorkResponse(*req.ID, true, nil)
		c.ch <- resp
		desc := fmt.Sprintf("submitted work %s from %s is not "+
			"less than the network target difficulty", hash.String(), id)
		return errs.PoolError(errs.NetworkDifficulty, desc)
	}
	headerB, err := header.Bytes()
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	submissionB := make([]byte, getworkDataLen)
	copy(submissionB[:wire.MaxBlockHeaderPayload], headerB)
	copy(submissionB[wire.MaxBlockHeaderPayload:],
		c.cfg.Blake256Pad)
	submission := hex.EncodeToString(submissionB)
	accepted, err := c.cfg.SubmitWork(ctx, &submission)
	if err != nil {
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	if !accepted {
		c.ch <- SubmitWorkResponse(*req.ID, false, nil)
		desc := fmt.Sprintf("%s: work %s rejected by the network",
			id, hash.String())
		if err != nil {
			if strings.Contains(err.Error(), "block difficulty of") {
				c.updateWork(true)
			}
			desc = fmt.Sprintf("%s: work %s rejected by the network (%v)",
				id, hash.String(), err)
		}
		return errs.PoolError(errs.WorkRejected, desc)
	}
	work := NewAcceptedWork(hash.String(), header.PrevBlock.String(),
		header.Height, c.account, miner)
	err = c.cfg.db.persistAcceptedWork(work)
	if err != nil {
		if errors.Is(err, errs.ValueFound) {
			sErr := NewStratumError(DuplicateShare, err)
			resp := SubmitWorkResponse(*req.ID, false, sErr)
			c.ch <- resp
			return err
		}
		sErr := NewStratumError(Unknown, err)
		resp := SubmitWorkResponse(*req.ID, false, sErr)
		c.ch <- resp
		return err
	}
	log.Tracef("Work %s accepted by the network", hash.String())
	resp := SubmitWorkResponse(*req.ID, true, nil)
	c.ch <- resp
	return nil
}