fix(reload): apply custom diff updates to active miners

Co-Authored-By: Virgil <virgil@lethean.io>

core_impl.go

@@ -296,7 +296,10 @@ func (cd *CustomDiff) OnLogin(e Event) {
 	if e.Miner.customDiffResolved {
 		return
 	}
-	e.Miner.user, e.Miner.customDiff = resolveLoginCustomDiff(e.Miner.user, cd.globalDiff.Load())
+	resolved := resolveLoginCustomDiff(e.Miner.user, cd.globalDiff.Load())
+	e.Miner.user = resolved.user
+	e.Miner.customDiff = resolved.diff
+	e.Miner.customDiffFromLogin = resolved.fromLogin
 	e.Miner.customDiffResolved = true
 }
 

miner.go

@@ -25,40 +25,41 @@ const (
 //	m := proxy.NewMiner(conn, 3333, nil)
 //	m.Start()
 type Miner struct {
-	id                 int64  // monotonically increasing per-process; atomic assignment
+	id                  int64  // monotonically increasing per-process; atomic assignment
-	rpcID              string // UUID v4 sent to miner as session id
+	rpcID               string // UUID v4 sent to miner as session id
-	state              MinerState
+	state               MinerState
-	extAlgo            bool // miner sent algo list in login params
+	extAlgo             bool // miner sent algo list in login params
-	loginAlgos         []string
+	loginAlgos          []string
-	extNH              bool   // NiceHash mode active (fixed byte splitting)
+	extNH               bool   // NiceHash mode active (fixed byte splitting)
-	algoEnabled        bool   // proxy is configured to negotiate the algo extension
+	algoEnabled         bool   // proxy is configured to negotiate the algo extension
-	ip                 string // remote IP (without port, for logging)
+	ip                  string // remote IP (without port, for logging)
-	remoteAddr         string
+	remoteAddr          string
-	localPort          uint16
+	localPort           uint16
-	user               string // login params.login (wallet address), custom diff suffix stripped
+	user                string // login params.login (wallet address), custom diff suffix stripped
-	password           string // login params.pass
+	password            string // login params.pass
-	agent              string // login params.agent
+	agent               string // login params.agent
-	rigID              string // login params.rigid (optional extension)
+	rigID               string // login params.rigid (optional extension)
-	fixedByte          uint8  // NiceHash slot index (0-255)
+	fixedByte           uint8  // NiceHash slot index (0-255)
-	mapperID           int64  // which NonceMapper owns this miner; -1 = unassigned
+	mapperID            int64  // which NonceMapper owns this miner; -1 = unassigned
-	routeID            int64  // SimpleMapper ID in simple mode; -1 = unassigned
+	routeID             int64  // SimpleMapper ID in simple mode; -1 = unassigned
-	customDiff         uint64 // 0 = use pool diff; non-zero = cap diff to this value
+	customDiff          uint64 // 0 = use pool diff; non-zero = cap diff to this value
-	customDiffResolved bool
+	customDiffResolved  bool
-	accessPassword     string
+	customDiffFromLogin bool
-	globalDiff         uint64
+	accessPassword      string
-	diff               uint64 // last difficulty sent to this miner from the pool
+	globalDiff          uint64
-	rx                 uint64 // bytes received from miner
+	diff                uint64 // last difficulty sent to this miner from the pool
-	tx                 uint64 // bytes sent from miner
+	rx                  uint64 // bytes received from miner
-	currentJob         Job
+	tx                  uint64 // bytes sent from miner
-	connectedAt        time.Time
+	currentJob          Job
-	lastActivityAt     time.Time
+	connectedAt         time.Time
-	conn               net.Conn
+	lastActivityAt      time.Time
-	tlsConn            *tls.Conn   // nil if plain TCP
+	conn                net.Conn
-	sendMu             sync.Mutex  // serialises writes to conn
+	tlsConn             *tls.Conn   // nil if plain TCP
-	buf                [16384]byte // per-miner send buffer; avoids per-write allocations
+	sendMu              sync.Mutex  // serialises writes to conn
-	onLogin            func(*Miner)
+	buf                 [16384]byte // per-miner send buffer; avoids per-write allocations
-	onLoginReady       func(*Miner)
+	onLogin             func(*Miner)
-	onSubmit           func(*Miner, *SubmitEvent)
+	onLoginReady        func(*Miner)
-	onClose            func(*Miner)
+	onSubmit            func(*Miner, *SubmitEvent)
-	closeOnce          sync.Once
+	onClose             func(*Miner)
+	closeOnce           sync.Once
 }

reload_test.go

@@ -1,6 +1,13 @@
 package proxy
 
-import "testing"
+import (
+	"bufio"
+	"encoding/json"
+	"net"
+	"strings"
+	"testing"
+	"time"
+)
 
 type reloadableSplitter struct {
 	reloads int
@@ -105,6 +112,160 @@ func TestProxy_Reload_WorkersMode_Good(t *testing.T) {
 	}
 }
 
+func TestProxy_Reload_CustomDiff_Good(t *testing.T) {
+	minerConn, clientConn := net.Pipe()
+	defer minerConn.Close()
+	defer clientConn.Close()
+
+	miner := NewMiner(minerConn, 3333, nil)
+	miner.state = MinerStateReady
+	miner.globalDiff = 1000
+	miner.customDiff = 1000
+	miner.currentJob = Job{
+		Blob:   strings.Repeat("0", 160),
+		JobID:  "job-1",
+		Target: "01000000",
+		Algo:   "cn/r",
+	}
+
+	p := &Proxy{
+		config: &Config{
+			Mode:       "nicehash",
+			Workers:    WorkersByRigID,
+			Bind:       []BindAddr{{Host: "127.0.0.1", Port: 3333}},
+			Pools:      []PoolConfig{{URL: "pool.example:3333", Enabled: true}},
+			CustomDiff: 1000,
+		},
+		customDiff: NewCustomDiff(1000),
+		miners:     map[int64]*Miner{miner.ID(): miner},
+	}
+
+	done := make(chan map[string]any, 1)
+	go func() {
+		line, err := bufio.NewReader(clientConn).ReadBytes('\n')
+		if err != nil {
+			done <- nil
+			return
+		}
+		var payload map[string]any
+		if err := json.Unmarshal(line, &payload); err != nil {
+			done <- nil
+			return
+		}
+		done <- payload
+	}()
+
+	p.Reload(&Config{
+		Mode:       "nicehash",
+		Workers:    WorkersByRigID,
+		Bind:       []BindAddr{{Host: "127.0.0.1", Port: 3333}},
+		Pools:      []PoolConfig{{URL: "pool.example:3333", Enabled: true}},
+		CustomDiff: 5000,
+	})
+
+	select {
+	case payload := <-done:
+		if payload == nil {
+			t.Fatal("expected reload to resend the current job with the new custom diff")
+		}
+		params, ok := payload["params"].(map[string]any)
+		if !ok {
+			t.Fatalf("expected job params payload, got %#v", payload["params"])
+		}
+		target, _ := params["target"].(string)
+		if got := (Job{Target: target}).DifficultyFromTarget(); got == 0 || got > 5000 {
+			t.Fatalf("expected resent job difficulty at or below 5000, got %d", got)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for reload job refresh")
+	}
+
+	if miner.customDiff != 5000 {
+		t.Fatalf("expected active miner custom diff to reload, got %d", miner.customDiff)
+	}
+	if miner.globalDiff != 5000 {
+		t.Fatalf("expected active miner global diff to reload, got %d", miner.globalDiff)
+	}
+}
+
+func TestProxy_Reload_CustomDiff_Bad(t *testing.T) {
+	miner := &Miner{
+		id:                  9,
+		state:               MinerStateReady,
+		globalDiff:          1000,
+		customDiff:          7000,
+		customDiffFromLogin: true,
+		currentJob: Job{
+			Blob:   strings.Repeat("0", 160),
+			JobID:  "job-1",
+			Target: "01000000",
+		},
+	}
+
+	p := &Proxy{
+		config: &Config{
+			Mode:       "nicehash",
+			Workers:    WorkersByRigID,
+			Bind:       []BindAddr{{Host: "127.0.0.1", Port: 3333}},
+			Pools:      []PoolConfig{{URL: "pool.example:3333", Enabled: true}},
+			CustomDiff: 1000,
+		},
+		customDiff: NewCustomDiff(1000),
+		miners:     map[int64]*Miner{miner.ID(): miner},
+	}
+
+	p.Reload(&Config{
+		Mode:       "nicehash",
+		Workers:    WorkersByRigID,
+		Bind:       []BindAddr{{Host: "127.0.0.1", Port: 3333}},
+		Pools:      []PoolConfig{{URL: "pool.example:3333", Enabled: true}},
+		CustomDiff: 5000,
+	})
+
+	if miner.customDiff != 7000 {
+		t.Fatalf("expected login suffix custom diff to be preserved, got %d", miner.customDiff)
+	}
+	if miner.globalDiff != 5000 {
+		t.Fatalf("expected miner global diff to update for future logins, got %d", miner.globalDiff)
+	}
+}
+
+func TestProxy_Reload_CustomDiff_Ugly(t *testing.T) {
+	miner := &Miner{
+		id:         11,
+		state:      MinerStateWaitLogin,
+		globalDiff: 1000,
+		customDiff: 1000,
+	}
+
+	p := &Proxy{
+		config: &Config{
+			Mode:       "nicehash",
+			Workers:    WorkersByRigID,
+			Bind:       []BindAddr{{Host: "127.0.0.1", Port: 3333}},
+			Pools:      []PoolConfig{{URL: "pool.example:3333", Enabled: true}},
+			CustomDiff: 1000,
+		},
+		customDiff: NewCustomDiff(1000),
+		miners:     map[int64]*Miner{miner.ID(): miner},
+	}
+
+	p.Reload(&Config{
+		Mode:       "nicehash",
+		Workers:    WorkersByRigID,
+		Bind:       []BindAddr{{Host: "127.0.0.1", Port: 3333}},
+		Pools:      []PoolConfig{{URL: "pool.example:3333", Enabled: true}},
+		CustomDiff: 0,
+	})
+
+	if miner.customDiff != 0 {
+		t.Fatalf("expected reload to clear the global custom diff for unauthenticated miners, got %d", miner.customDiff)
+	}
+	if miner.globalDiff != 0 {
+		t.Fatalf("expected miner global diff to be cleared, got %d", miner.globalDiff)
+	}
+}
+
 func TestProxy_Reload_UpdatesServers(t *testing.T) {
 	originalLimiter := NewRateLimiter(RateLimit{MaxConnectionsPerMinute: 1})
 	p := &Proxy{

state_impl.go

@@ -375,6 +375,7 @@ func (p *Proxy) Reload(config *Config) {
 	if p.customDiff != nil {
 		p.customDiff.globalDiff.Store(config.CustomDiff)
 	}
+	p.reloadCustomDiff(config.CustomDiff)
 	if p.customDiffBuckets != nil {
 		p.customDiffBuckets.SetEnabled(config.CustomDiffStats)
 	}
@@ -398,6 +399,29 @@ func (p *Proxy) Reload(config *Config) {
 	}
 }
 
+func (p *Proxy) reloadCustomDiff(globalDiff uint64) {
+	if p == nil {
+		return
+	}
+	for _, miner := range p.activeMiners() {
+		if miner == nil {
+			continue
+		}
+		miner.globalDiff = globalDiff
+		if miner.customDiffFromLogin {
+			continue
+		}
+		miner.customDiff = globalDiff
+		switch miner.state {
+		case MinerStateWaitReady, MinerStateReady:
+			job := miner.CurrentJob()
+			if job.IsValid() {
+				miner.ForwardJob(job, job.Algo)
+			}
+		}
+	}
+}
+
 func (p *Proxy) reloadWatcher(enabled bool) {
 	if p == nil || p.config == nil || p.config.configPath == "" {
 		return
@@ -1010,7 +1034,10 @@ func (m *Miner) handleLogin(request stratumRequest) {
 		m.ReplyWithError(requestID(request.ID), "Invalid password")
 		return
 	}
-	m.user, m.customDiff = resolveLoginCustomDiff(params.Login, m.globalDiff)
+	resolved := resolveLoginCustomDiff(params.Login, m.globalDiff)
+	m.user = resolved.user
+	m.customDiff = resolved.diff
+	m.customDiffFromLogin = resolved.fromLogin
 	m.customDiffResolved = true
 	m.password = params.Pass
 	m.agent = params.Agent
@@ -1044,21 +1071,27 @@ func (m *Miner) handleLogin(request stratumRequest) {
 	m.replyLoginSuccess(requestID(request.ID))
 }
 
-func resolveLoginCustomDiff(login string, globalDiff uint64) (string, uint64) {
+type resolvedCustomDiff struct {
+	user      string
+	diff      uint64
+	fromLogin bool
+}
+
+func resolveLoginCustomDiff(login string, globalDiff uint64) resolvedCustomDiff {
 	plus := strings.LastIndex(login, "+")
 	if plus >= 0 && plus < len(login)-1 {
 		suffix := login[plus+1:]
 		if isDecimalDigits(suffix) {
 			if parsed, err := strconv.ParseUint(suffix, 10, 64); err == nil {
-				return login[:plus], parsed
+				return resolvedCustomDiff{user: login[:plus], diff: parsed, fromLogin: true}
 			}
 		}
-		return login, 0
+		return resolvedCustomDiff{user: login}
 	}
 	if globalDiff > 0 {
-		return login, globalDiff
+		return resolvedCustomDiff{user: login, diff: globalDiff}
 	}
-	return login, 0
+	return resolvedCustomDiff{user: login}
 }
 
 func isDecimalDigits(value string) bool {