go-proxy/miner_runtime.go

package proxy

import (
	"bufio"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"net"
	"strings"
	"time"
)

type minerRequest struct {
	ID     int64           `json:"id"`
	Method string          `json:"method"`
	Params json.RawMessage `json:"params"`
}

// Start begins the read loop in a goroutine and arms the login timeout timer.
//
//	m.Start()
func (m *Miner) Start() {
	if m == nil || m.conn == nil {
		return
	}

	go func() {
		reader := bufio.NewReaderSize(m.conn, len(m.buf))
		for {
			if errorValue := m.applyReadDeadline(); errorValue != nil {
				m.Close()
				return
			}
			line, isPrefix, errorValue := reader.ReadLine()
			if errorValue != nil {
				m.Close()
				return
			}
			if isPrefix {
				m.Close()
				return
			}
			if len(line) == 0 {
				continue
			}
			m.rx += uint64(len(line) + 1)
			m.Touch()
			m.handleLine(line)
		}
	}()
}

// ForwardJob encodes the job as a stratum job notification and writes it to the miner.
//
//	m.ForwardJob(job, "cn/r")
func (m *Miner) ForwardJob(job Job, algo string) {
	if m == nil || m.conn == nil {
		return
	}

	blob := job.Blob
	if m.extNH {
		blob = job.BlobWithFixedByte(m.fixedByte)
	}

	m.diff = job.DifficultyFromTarget()
	m.SetState(MinerStateReady)
	jobCopy := job
	m.currentJob = &jobCopy
	m.Touch()

	params := map[string]interface{}{
		"blob":   blob,
		"job_id": job.JobID,
		"target": job.Target,
		"id":     m.rpcID,
	}
	if m.algoExtension && m.extAlgo && algo != "" {
		params["algo"] = algo
	}

	m.writeJSON(map[string]interface{}{
		"jsonrpc": "2.0",
		"method":  "job",
		"params":  params,
	})
}

// ReplyWithError sends a JSON-RPC error response for the given request id.
//
//	m.ReplyWithError(2, "Low difficulty share")
func (m *Miner) ReplyWithError(id int64, message string) {
	m.writeJSON(map[string]interface{}{
		"id":      id,
		"jsonrpc": "2.0",
		"error": map[string]interface{}{
			"code":    -1,
			"message": message,
		},
	})
}

// Success sends a JSON-RPC success response with the given status string.
//
//	m.Success(2, "OK")
func (m *Miner) Success(id int64, status string) {
	m.writeJSON(map[string]interface{}{
		"id":      id,
		"jsonrpc": "2.0",
		"error":   nil,
		"result": map[string]string{
			"status": status,
		},
	})
}

// Close initiates graceful TCP shutdown. Safe to call multiple times.
//
//	m.Close()
func (m *Miner) Close() {
	if m == nil || m.conn == nil {
		return
	}

	m.closeOnce.Do(func() {
		m.SetState(MinerStateClosing)
		if m.events != nil {
			m.events.Dispatch(Event{Type: EventClose, Miner: m})
		}
		_ = m.conn.Close()
	})
}

func (m *Miner) writeJSON(value interface{}) {
	if m == nil || m.conn == nil {
		return
	}

	data, errorValue := json.Marshal(value)
	if errorValue != nil {
		return
	}

	m.sendMu.Lock()
	defer m.sendMu.Unlock()

	data = append(data, '\n')
	written, errorValue := m.conn.Write(data)
	if errorValue == nil {
		m.tx += uint64(written)
	}
}

func (m *Miner) handleLine(line []byte) {
	if len(line) > len(m.buf) {
		m.Close()
		return
	}

	request := minerRequest{}
	if errorValue := json.Unmarshal(line, &request); errorValue != nil {
		m.Close()
		return
	}

	switch request.Method {
	case "login":
		m.handleLogin(request)
	case "submit":
		m.handleSubmit(request)
	case "keepalived":
		m.handleKeepalived(request)
	default:
		if request.ID != 0 {
			m.ReplyWithError(request.ID, "Invalid request")
		}
	}
}

func (m *Miner) handleLogin(request minerRequest) {
	type loginParams struct {
		Login string   `json:"login"`
		Pass  string   `json:"pass"`
		Agent string   `json:"agent"`
		Algo  []string `json:"algo"`
		RigID string   `json:"rigid"`
	}

	params := loginParams{}
	if errorValue := json.Unmarshal(request.Params, &params); errorValue != nil {
		m.ReplyWithError(request.ID, "Invalid payment address provided")
		return
	}

	if params.Login == "" {
		m.ReplyWithError(request.ID, "Invalid payment address provided")
		return
	}
	if m.accessPassword != "" && params.Pass != m.accessPassword {
		m.ReplyWithError(request.ID, "Invalid password")
		return
	}

	m.SetPassword(params.Pass)
	m.SetAgent(params.Agent)
	m.SetRigID(params.RigID)
	m.algo = append(m.algo[:0], params.Algo...)
	m.extAlgo = len(params.Algo) > 0
	m.SetUser(params.Login)
	m.SetRPCID(newRPCID())

	if m.events != nil {
		m.events.Dispatch(Event{Type: EventLogin, Miner: m})
	}

	m.Touch()
	if m.State() == MinerStateClosing {
		return
	}

	result := map[string]interface{}{
		"id":     m.rpcID,
		"status": "OK",
	}
	if m.currentJob != nil && m.currentJob.IsValid() {
		jobCopy := *m.currentJob
		blob := jobCopy.Blob
		if m.extNH {
			blob = jobCopy.BlobWithFixedByte(m.fixedByte)
		}
		jobResult := map[string]interface{}{
			"blob":   blob,
			"job_id": jobCopy.JobID,
			"target": jobCopy.Target,
			"id":     m.rpcID,
		}
		if m.algoExtension && m.extAlgo && jobCopy.Algo != "" {
			jobResult["algo"] = jobCopy.Algo
		}
		result["job"] = jobResult
		if m.algoExtension && m.extAlgo {
			result["extensions"] = []string{"algo"}
		}
		m.SetState(MinerStateReady)
	} else {
		m.SetState(MinerStateWaitReady)
		if m.algoExtension && m.extAlgo {
			result["extensions"] = []string{"algo"}
		}
	}

	m.writeJSON(map[string]interface{}{
		"id":      request.ID,
		"jsonrpc": "2.0",
		"error":   nil,
		"result":  result,
	})
}

func (m *Miner) handleSubmit(request minerRequest) {
	if m.State() != MinerStateReady {
		m.ReplyWithError(request.ID, "Unauthenticated")
		return
	}

	type submitParams struct {
		ID     string `json:"id"`
		JobID  string `json:"job_id"`
		Nonce  string `json:"nonce"`
		Result string `json:"result"`
		Algo   string `json:"algo"`
	}

	params := submitParams{}
	if errorValue := json.Unmarshal(request.Params, &params); errorValue != nil {
		m.ReplyWithError(request.ID, "Malformed share")
		return
	}

	if params.ID != m.rpcID {
		m.ReplyWithError(request.ID, "Unauthenticated")
		return
	}
	if params.JobID == "" {
		m.ReplyWithError(request.ID, "Missing job id")
		return
	}
	if len(params.Nonce) != 8 || params.Nonce != strings.ToLower(params.Nonce) {
		m.ReplyWithError(request.ID, "Invalid nonce")
		return
	}
	if _, errorValue := hex.DecodeString(params.Nonce); errorValue != nil {
		m.ReplyWithError(request.ID, "Invalid nonce")
		return
	}

	submitAlgo := ""
	if m.algoExtension && m.extAlgo {
		submitAlgo = params.Algo
	}

	m.Touch()
	if m.events != nil {
		m.events.Dispatch(Event{
			Type:      EventSubmit,
			Miner:     m,
			JobID:     params.JobID,
			Nonce:     params.Nonce,
			Result:    params.Result,
			Algo:      submitAlgo,
			RequestID: request.ID,
		})
		return
	}

	if m.splitter != nil {
		m.splitter.OnSubmit(&SubmitEvent{
			Miner:     m,
			JobID:     params.JobID,
			Nonce:     params.Nonce,
			Result:    params.Result,
			Algo:      submitAlgo,
			RequestID: request.ID,
		})
	}
}

func (m *Miner) handleKeepalived(request minerRequest) {
	m.Touch()
	m.Success(request.ID, "KEEPALIVED")
}

func (m *Miner) currentJobCopy() *Job {
	if m == nil || m.currentJob == nil {
		return nil
	}

	jobCopy := *m.currentJob
	return &jobCopy
}

func (m *Miner) applyReadDeadline() error {
	if m == nil || m.conn == nil {
		return nil
	}

	deadline := m.readDeadline()
	if deadline.IsZero() {
		return nil
	}

	return m.conn.SetReadDeadline(deadline)
}

func (m *Miner) readDeadline() time.Time {
	if m == nil {
		return time.Time{}
	}

	switch m.State() {
	case MinerStateWaitLogin:
		return m.lastActivityAt.Add(10 * time.Second)
	case MinerStateWaitReady, MinerStateReady:
		return m.lastActivityAt.Add(600 * time.Second)
	default:
		return time.Time{}
	}
}

func (m *Miner) dispatchSubmitResult(eventType EventType, diff uint64, errorMessage string, requestID int64) {
	if m == nil || m.events == nil {
		return
	}

	jobCopy := m.currentJobCopy()
	m.events.Dispatch(Event{
		Type:    eventType,
		Miner:   m,
		Job:     jobCopy,
		Diff:    diff,
		Error:   errorMessage,
		Latency: 0,
	})
	if eventType == EventAccept {
		m.Success(requestID, "OK")
		return
	}
	m.ReplyWithError(requestID, errorMessage)
}

func (m *Miner) setStateFromJob(job Job) {
	m.currentJob = &job
	m.SetState(MinerStateReady)
}

func (m *Miner) Expire() {
	if m == nil || m.State() == MinerStateClosing {
		return
	}
	m.Close()
}

func newRPCID() string {
	value := make([]byte, 16)
	_, _ = rand.Read(value)
	value[6] = (value[6] & 0x0f) | 0x40
	value[8] = (value[8] & 0x3f) | 0x80

	encoded := make([]byte, 36)
	hex.Encode(encoded[0:8], value[0:4])
	encoded[8] = '-'
	hex.Encode(encoded[9:13], value[4:6])
	encoded[13] = '-'
	hex.Encode(encoded[14:18], value[6:8])
	encoded[18] = '-'
	hex.Encode(encoded[19:23], value[8:10])
	encoded[23] = '-'
	hex.Encode(encoded[24:36], value[10:16])
	return string(encoded)
}

func (m *Miner) RemoteAddr() net.Addr {
	if m == nil || m.conn == nil {
		return nil
	}
	return m.conn.RemoteAddr()
}