Mining/pkg/mining/manager.go
Claude 0d1b20e177
ax(batch): replace prose comments with usage examples across all packages
Applies AX principle 2 (Comments as Usage Examples) — removes prose
descriptions that restate the function signature ("returns", "retrieves",
"creates", "wraps", etc.) and keeps or replaces with concrete usage
examples showing real calls with realistic values.

Co-Authored-By: Charon <charon@lethean.io>
2026-04-02 18:28:16 +01:00

812 lines
23 KiB
Go

package mining
import (
"bytes"
"context"
"net"
"regexp"
"strconv"
"sync"
"time"
"forge.lthn.ai/Snider/Mining/pkg/database"
"forge.lthn.ai/Snider/Mining/pkg/logging"
)
// equalFold("xmrig", "XMRig") == true
// equalFold("tt-miner", "TT-Miner") == true
func equalFold(left, right string) bool {
return bytes.EqualFold([]byte(left), []byte(right))
}
// hasPrefix("xmrig-rx0", "xmrig") == true
// hasPrefix("ttminer-rtx", "xmrig") == false
func hasPrefix(input, prefix string) bool {
return len(input) >= len(prefix) && input[:len(prefix)] == prefix
}
// containsStr("peer not found", "not found") == true
// containsStr("connection ok", "not found") == false
func containsStr(haystack, needle string) bool {
if len(needle) == 0 {
return true
}
if len(haystack) < len(needle) {
return false
}
for i := 0; i <= len(haystack)-len(needle); i++ {
if haystack[i:i+len(needle)] == needle {
return true
}
}
return false
}
// safe := instanceNameRegex.ReplaceAllString("my algo!", "_") // => "my_algo_"
var instanceNameRegex = regexp.MustCompile(`[^a-zA-Z0-9_/-]`)
// var managerInterface ManagerInterface = mining.NewManager()
// miner, err := managerInterface.StartMiner(ctx, "xmrig", &mining.Config{Algo: "rx/0"})
// defer managerInterface.Stop()
type ManagerInterface interface {
StartMiner(ctx context.Context, minerType string, config *Config) (Miner, error)
StopMiner(ctx context.Context, name string) error
GetMiner(name string) (Miner, error)
ListMiners() []Miner
ListAvailableMiners() []AvailableMiner
GetMinerHashrateHistory(name string) ([]HashratePoint, error)
UninstallMiner(ctx context.Context, minerType string) error
Stop()
}
// manager := mining.NewManager()
// defer manager.Stop()
// miner, err := manager.StartMiner(ctx, "xmrig", &mining.Config{Algo: "rx/0"})
type Manager struct {
miners map[string]Miner
mutex sync.RWMutex
stopChan chan struct{}
stopOnce sync.Once
waitGroup sync.WaitGroup
databaseEnabled bool
databaseRetention int
eventHub *EventHub
eventHubMutex sync.RWMutex // Separate mutex for eventHub to avoid deadlock with main mutex
}
// m.SetEventHub(eventHub)
func (m *Manager) SetEventHub(hub *EventHub) {
m.eventHubMutex.Lock()
defer m.eventHubMutex.Unlock()
m.eventHub = hub
}
// m.emitEvent(EventMinerStarted, MinerEventData{Name: instanceName})
// m.emitEvent(EventMinerError, MinerEventData{Name: instanceName, Error: err.Error()})
func (m *Manager) emitEvent(eventType EventType, data interface{}) {
m.eventHubMutex.RLock()
hub := m.eventHub
m.eventHubMutex.RUnlock()
if hub != nil {
hub.Broadcast(NewEvent(eventType, data))
}
}
var _ ManagerInterface = (*Manager)(nil)
// manager := mining.NewManager()
// defer manager.Stop()
func NewManager() *Manager {
m := &Manager{
miners: make(map[string]Miner),
stopChan: make(chan struct{}),
waitGroup: sync.WaitGroup{},
}
m.syncMinersConfig() // Ensure config file is populated
m.initDatabase()
m.autostartMiners()
m.startStatsCollection()
return m
}
// manager := mining.NewManagerForSimulation()
// manager.StartMiner(ctx, "xmrig", &mining.Config{Algo: "rx/0"})
func NewManagerForSimulation() *Manager {
m := &Manager{
miners: make(map[string]Miner),
stopChan: make(chan struct{}),
waitGroup: sync.WaitGroup{},
}
// Skip syncMinersConfig and autostartMiners for simulation
m.startStatsCollection()
return m
}
// m.initDatabase() // called in NewManager(); enables persistence if MinersConfig.Database.Enabled == true
func (m *Manager) initDatabase() {
minersConfiguration, err := LoadMinersConfig()
if err != nil {
logging.Warn("could not load config for database init", logging.Fields{"error": err})
return
}
m.databaseEnabled = minersConfiguration.Database.Enabled
m.databaseRetention = minersConfiguration.Database.RetentionDays
if m.databaseRetention == 0 {
m.databaseRetention = 30
}
if !m.databaseEnabled {
logging.Debug("database persistence is disabled")
return
}
databaseConfiguration := database.Config{
Enabled: true,
RetentionDays: m.databaseRetention,
}
if err := database.Initialize(databaseConfiguration); err != nil {
logging.Warn("failed to initialize database", logging.Fields{"error": err})
m.databaseEnabled = false
return
}
logging.Info("database persistence enabled", logging.Fields{"retention_days": m.databaseRetention})
// Start periodic cleanup
m.startDBCleanup()
}
// m.startDBCleanup() // called after initDatabase(); purges rows older than m.databaseRetention days once per hour
func (m *Manager) startDBCleanup() {
m.waitGroup.Add(1)
go func() {
defer m.waitGroup.Done()
defer func() {
if r := recover(); r != nil {
logging.Error("panic in database cleanup goroutine", logging.Fields{"panic": r})
}
}()
// Run cleanup once per hour
ticker := time.NewTicker(time.Hour)
defer ticker.Stop()
// Run initial cleanup
if err := database.Cleanup(m.databaseRetention); err != nil {
logging.Warn("database cleanup failed", logging.Fields{"error": err})
}
for {
select {
case <-ticker.C:
if err := database.Cleanup(m.databaseRetention); err != nil {
logging.Warn("database cleanup failed", logging.Fields{"error": err})
}
case <-m.stopChan:
return
}
}
}()
}
// m.syncMinersConfig() // called on startup; adds {MinerType: "xmrig", Autostart: false} for any
// registered miner not yet present in miners.json — existing entries are left unchanged.
func (m *Manager) syncMinersConfig() {
minersConfiguration, err := LoadMinersConfig()
if err != nil {
logging.Warn("could not load miners config for sync", logging.Fields{"error": err})
return
}
availableMiners := m.ListAvailableMiners()
configUpdated := false
for _, availableMiner := range availableMiners {
found := false
for _, configuredMiner := range minersConfiguration.Miners {
if equalFold(configuredMiner.MinerType, availableMiner.Name) {
found = true
break
}
}
if !found {
minersConfiguration.Miners = append(minersConfiguration.Miners, MinerAutostartConfig{
MinerType: availableMiner.Name,
Autostart: false,
Config: nil, // No default config
})
configUpdated = true
logging.Info("added default config for missing miner", logging.Fields{"miner": availableMiner.Name})
}
}
if configUpdated {
if err := SaveMinersConfig(minersConfiguration); err != nil {
logging.Warn("failed to save updated miners config", logging.Fields{"error": err})
}
}
}
// m.autostartMiners() // called in NewManager(); reads miners.json and starts any entry with Autostart == true
func (m *Manager) autostartMiners() {
minersConfiguration, err := LoadMinersConfig()
if err != nil {
logging.Warn("could not load miners config for autostart", logging.Fields{"error": err})
return
}
for _, autostartEntry := range minersConfiguration.Miners {
if autostartEntry.Autostart && autostartEntry.Config != nil {
logging.Info("autostarting miner", logging.Fields{"type": autostartEntry.MinerType})
if _, err := m.StartMiner(context.Background(), autostartEntry.MinerType, autostartEntry.Config); err != nil {
logging.Error("failed to autostart miner", logging.Fields{"type": autostartEntry.MinerType, "error": err})
}
}
}
}
// port, err := findAvailablePort()
// if err != nil { return 0, err }
// config.HTTPPort = port
func findAvailablePort() (int, error) {
addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
if err != nil {
return 0, err
}
listener, err := net.ListenTCP("tcp", addr)
if err != nil {
return 0, err
}
defer listener.Close()
return listener.Addr().(*net.TCPAddr).Port, nil
}
// miner, err := manager.StartMiner(ctx, "xmrig", &mining.Config{Algo: "rx/0"})
func (m *Manager) StartMiner(ctx context.Context, minerType string, config *Config) (Miner, error) {
// Check for cancellation before acquiring lock
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
m.mutex.Lock()
defer m.mutex.Unlock()
if config == nil {
config = &Config{}
}
miner, err := CreateMiner(minerType)
if err != nil {
return nil, err
}
instanceName := miner.GetName()
if config.Algo != "" {
// Sanitize algo to prevent directory traversal or invalid filenames
sanitizedAlgo := instanceNameRegex.ReplaceAllString(config.Algo, "_")
instanceName = instanceName + "-" + sanitizedAlgo
} else {
instanceName = instanceName + "-" + strconv.FormatInt(time.Now().UnixNano()%1000, 10)
}
if _, exists := m.miners[instanceName]; exists {
return nil, ErrMinerExists(instanceName)
}
// Validate user-provided HTTPPort if specified
if config.HTTPPort != 0 {
if config.HTTPPort < 1024 || config.HTTPPort > 65535 {
return nil, ErrInvalidConfig("HTTPPort must be between 1024 and 65535, got " + strconv.Itoa(config.HTTPPort))
}
}
apiPort, err := findAvailablePort()
if err != nil {
return nil, ErrInternal("failed to find an available port for the miner API").WithCause(err)
}
if config.HTTPPort == 0 {
config.HTTPPort = apiPort
}
if xmrigMiner, ok := miner.(*XMRigMiner); ok {
xmrigMiner.Name = instanceName
if xmrigMiner.API != nil {
xmrigMiner.API.ListenPort = apiPort
}
}
if ttMiner, ok := miner.(*TTMiner); ok {
ttMiner.Name = instanceName
if ttMiner.API != nil {
ttMiner.API.ListenPort = apiPort
}
}
// Emit starting event before actually starting
m.emitEvent(EventMinerStarting, MinerEventData{
Name: instanceName,
})
if err := miner.Start(config); err != nil {
// Emit error event
m.emitEvent(EventMinerError, MinerEventData{
Name: instanceName,
Error: err.Error(),
})
return nil, err
}
m.miners[instanceName] = miner
if err := m.updateMinerConfig(minerType, true, config); err != nil {
logging.Warn("failed to save miner config for autostart", logging.Fields{"error": err})
}
logMessage := "CryptoCurrency Miner started: " + miner.GetName() + " (Binary: " + miner.GetBinaryPath() + ")"
logToSyslog(logMessage)
// Emit started event
m.emitEvent(EventMinerStarted, MinerEventData{
Name: instanceName,
})
RecordMinerStart()
return miner, nil
}
// manager.UninstallMiner(ctx, "xmrig") // stops all xmrig instances and removes config
// manager.UninstallMiner(ctx, "ttminer") // stops all ttminer instances and removes config
func (m *Manager) UninstallMiner(ctx context.Context, minerType string) error {
// Check for cancellation before acquiring lock
select {
case <-ctx.Done():
return ctx.Err()
default:
}
m.mutex.Lock()
// Collect miners to stop and delete (can't modify map during iteration)
minersToDelete := make([]string, 0)
minersToStop := make([]Miner, 0)
for name, runningMiner := range m.miners {
if rm, ok := runningMiner.(*XMRigMiner); ok && equalFold(rm.ExecutableName, minerType) {
minersToStop = append(minersToStop, runningMiner)
minersToDelete = append(minersToDelete, name)
}
if rm, ok := runningMiner.(*TTMiner); ok && equalFold(rm.ExecutableName, minerType) {
minersToStop = append(minersToStop, runningMiner)
minersToDelete = append(minersToDelete, name)
}
}
// Delete from map first, then release lock before stopping (Stop may block)
for _, name := range minersToDelete {
delete(m.miners, name)
}
m.mutex.Unlock()
// Stop miners outside the lock to avoid blocking
for i, miner := range minersToStop {
if err := miner.Stop(); err != nil {
logging.Warn("failed to stop running miner during uninstall", logging.Fields{"miner": minersToDelete[i], "error": err})
}
}
miner, err := CreateMiner(minerType)
if err != nil {
return err
}
if err := miner.Uninstall(); err != nil {
return ErrInternal("failed to uninstall miner files").WithCause(err)
}
return UpdateMinersConfig(func(configuration *MinersConfig) error {
var updatedMiners []MinerAutostartConfig
for _, autostartEntry := range configuration.Miners {
if !equalFold(autostartEntry.MinerType, minerType) {
updatedMiners = append(updatedMiners, autostartEntry)
}
}
configuration.Miners = updatedMiners
return nil
})
}
// m.updateMinerConfig("xmrig", true, config) // persists autostart=true and last-used config to miners.json
func (m *Manager) updateMinerConfig(minerType string, autostart bool, config *Config) error {
return UpdateMinersConfig(func(configuration *MinersConfig) error {
found := false
for i, autostartEntry := range configuration.Miners {
if equalFold(autostartEntry.MinerType, minerType) {
configuration.Miners[i].Autostart = autostart
configuration.Miners[i].Config = config
found = true
break
}
}
if !found {
configuration.Miners = append(configuration.Miners, MinerAutostartConfig{
MinerType: minerType,
Autostart: autostart,
Config: config,
})
}
return nil
})
}
// manager.StopMiner(ctx, "xmrig/monero")
// manager.StopMiner(ctx, "ttminer/rtx4090") // still removes if already stopped
func (m *Manager) StopMiner(ctx context.Context, name string) error {
// Check for cancellation before acquiring lock
select {
case <-ctx.Done():
return ctx.Err()
default:
}
m.mutex.Lock()
defer m.mutex.Unlock()
miner, exists := m.miners[name]
if !exists {
for k := range m.miners {
if hasPrefix(k, name) {
miner = m.miners[k]
name = k
exists = true
break
}
}
}
if !exists {
return ErrMinerNotFound(name)
}
// Emit stopping event
m.emitEvent(EventMinerStopping, MinerEventData{
Name: name,
})
// Try to stop the miner, but always remove it from the map
// This handles the case where a miner crashed or was killed externally
stopErr := miner.Stop()
// Always remove from map - if it's not running, we still want to clean it up
delete(m.miners, name)
// Emit stopped event
reason := "stopped"
if stopErr != nil && stopErr.Error() != "miner is not running" {
reason = stopErr.Error()
}
m.emitEvent(EventMinerStopped, MinerEventData{
Name: name,
Reason: reason,
})
// Only return error if it wasn't just "miner is not running"
if stopErr != nil && stopErr.Error() != "miner is not running" {
return stopErr
}
RecordMinerStop()
return nil
}
// miner, err := m.GetMiner("xmrig-randomx")
// if err != nil { /* miner not found */ }
func (m *Manager) GetMiner(name string) (Miner, error) {
m.mutex.RLock()
defer m.mutex.RUnlock()
miner, exists := m.miners[name]
if !exists {
return nil, ErrMinerNotFound(name)
}
return miner, nil
}
// miners := m.ListMiners()
// for _, miner := range miners { logging.Info(miner.GetName()) }
func (m *Manager) ListMiners() []Miner {
m.mutex.RLock()
defer m.mutex.RUnlock()
miners := make([]Miner, 0, len(m.miners))
for _, miner := range m.miners {
miners = append(miners, miner)
}
return miners
}
// sim := NewSimulatedMiner(SimulatedMinerConfig{Name: "sim-rx0"})
// if err := manager.RegisterMiner(sim); err != nil { return err }
func (m *Manager) RegisterMiner(miner Miner) error {
name := miner.GetName()
m.mutex.Lock()
if _, exists := m.miners[name]; exists {
m.mutex.Unlock()
return ErrMinerExists(name)
}
m.miners[name] = miner
m.mutex.Unlock()
logging.Info("registered miner", logging.Fields{"name": name})
// Emit miner started event (outside lock)
m.emitEvent(EventMinerStarted, map[string]interface{}{
"name": name,
})
return nil
}
// for _, available := range manager.ListAvailableMiners() { logging.Info(available.Name, nil) }
func (m *Manager) ListAvailableMiners() []AvailableMiner {
return []AvailableMiner{
{
Name: "xmrig",
Description: "XMRig is a high performance, open source, cross platform RandomX, KawPow, CryptoNight and AstroBWT CPU/GPU miner and RandomX benchmark.",
},
{
Name: "tt-miner",
Description: "TT-Miner is a high performance NVIDIA GPU miner for various algorithms including Ethash, KawPow, ProgPow, and more. Requires CUDA.",
},
}
}
// m.startStatsCollection() // called in NewManager(); polls all active miners every HighResolutionInterval
func (m *Manager) startStatsCollection() {
m.waitGroup.Add(1)
go func() {
defer m.waitGroup.Done()
defer func() {
if r := recover(); r != nil {
logging.Error("panic in stats collection goroutine", logging.Fields{"panic": r})
}
}()
ticker := time.NewTicker(HighResolutionInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
m.collectMinerStats()
case <-m.stopChan:
return
}
}
}()
}
// ctx, cancel := context.WithTimeout(ctx, statsCollectionTimeout)
const statsCollectionTimeout = 5 * time.Second
// m.collectMinerStats() // called by startStatsCollection ticker; gathers stats from all active miners in parallel
func (m *Manager) collectMinerStats() {
// Take a snapshot of miners under read lock - minimize lock duration
m.mutex.RLock()
if len(m.miners) == 0 {
m.mutex.RUnlock()
return
}
type minerInfo struct {
miner Miner
minerType string
}
miners := make([]minerInfo, 0, len(m.miners))
for _, miner := range m.miners {
// Use the miner's GetType() method for proper type identification
miners = append(miners, minerInfo{miner: miner, minerType: miner.GetType()})
}
databaseEnabled := m.databaseEnabled // Copy to avoid holding lock
m.mutex.RUnlock()
now := time.Now()
// Collect stats from all miners in parallel
var waitGroup sync.WaitGroup
for _, entry := range miners {
waitGroup.Add(1)
go func(miner Miner, minerType string) {
defer waitGroup.Done()
defer func() {
if r := recover(); r != nil {
logging.Error("panic in single miner stats collection", logging.Fields{
"panic": r,
"miner": miner.GetName(),
})
}
}()
m.collectSingleMinerStats(miner, minerType, now, databaseEnabled)
}(entry.miner, entry.minerType)
}
waitGroup.Wait()
}
// for attempt := 0; attempt <= statsRetryCount; attempt++ { ... }
const statsRetryCount = 2
// time.Sleep(statsRetryDelay) // between retry attempts
const statsRetryDelay = 500 * time.Millisecond
// m.collectSingleMinerStats(miner, "xmrig", time.Now(), true) // retries up to statsRetryCount times; persists to DB if databaseEnabled
func (m *Manager) collectSingleMinerStats(miner Miner, minerType string, now time.Time, databaseEnabled bool) {
minerName := miner.GetName()
var stats *PerformanceMetrics
var lastErr error
// Retry loop for transient failures
for attempt := 0; attempt <= statsRetryCount; attempt++ {
// Use context with timeout to prevent hanging on unresponsive miner APIs
ctx, cancel := context.WithTimeout(context.Background(), statsCollectionTimeout)
stats, lastErr = miner.GetStats(ctx)
cancel() // Release context immediately
if lastErr == nil {
break // Success
}
// Log retry attempts at debug level
if attempt < statsRetryCount {
logging.Debug("retrying stats collection", logging.Fields{
"miner": minerName,
"attempt": attempt + 1,
"error": lastErr.Error(),
})
time.Sleep(statsRetryDelay)
}
}
if lastErr != nil {
logging.Error("failed to get miner stats after retries", logging.Fields{
"miner": minerName,
"error": lastErr.Error(),
"retries": statsRetryCount,
})
RecordStatsCollection(true, true)
return
}
// Record stats collection (retried if we did any retries)
RecordStatsCollection(stats != nil && lastErr == nil, false)
point := HashratePoint{
Timestamp: now,
Hashrate: stats.Hashrate,
}
// Add to in-memory history (rolling window)
// Note: AddHashratePoint and ReduceHashrateHistory must be thread-safe
miner.AddHashratePoint(point)
miner.ReduceHashrateHistory(now)
// Persist to database if enabled
if databaseEnabled {
databasePoint := database.HashratePoint{
Timestamp: point.Timestamp,
Hashrate: point.Hashrate,
}
// Create a new context for DB writes (original context is from retry loop)
databaseContext, databaseCancel := context.WithTimeout(context.Background(), statsCollectionTimeout)
if err := database.InsertHashratePoint(databaseContext, minerName, minerType, databasePoint, database.ResolutionHigh); err != nil {
logging.Warn("failed to persist hashrate", logging.Fields{"miner": minerName, "error": err})
}
databaseCancel()
}
// Emit stats event for real-time WebSocket updates
m.emitEvent(EventMinerStats, MinerStatsData{
Name: minerName,
Hashrate: stats.Hashrate,
Shares: stats.Shares,
Rejected: stats.Rejected,
Uptime: stats.Uptime,
Algorithm: stats.Algorithm,
DiffCurrent: stats.DiffCurrent,
})
}
// points, err := manager.GetMinerHashrateHistory("xmrig")
// for _, point := range points { logging.Info("hashrate", logging.Fields{"time": point.Timestamp, "rate": point.Hashrate}) }
func (m *Manager) GetMinerHashrateHistory(name string) ([]HashratePoint, error) {
m.mutex.RLock()
defer m.mutex.RUnlock()
miner, exists := m.miners[name]
if !exists {
return nil, ErrMinerNotFound(name)
}
return miner.GetHashrateHistory(), nil
}
// ctx, cancel := context.WithTimeout(context.Background(), ShutdownTimeout)
const ShutdownTimeout = 10 * time.Second
// defer manager.Stop() // safe in main() or test cleanup; subsequent calls are no-ops
func (m *Manager) Stop() {
m.stopOnce.Do(func() {
// Stop all running miners first
m.mutex.Lock()
for name, miner := range m.miners {
if err := miner.Stop(); err != nil {
logging.Warn("failed to stop miner", logging.Fields{"miner": name, "error": err})
}
}
m.mutex.Unlock()
close(m.stopChan)
// Wait for goroutines with timeout
done := make(chan struct{})
go func() {
m.waitGroup.Wait()
close(done)
}()
select {
case <-done:
logging.Info("all goroutines stopped gracefully")
case <-time.After(ShutdownTimeout):
logging.Warn("shutdown timeout - some goroutines may not have stopped")
}
// Close the database
if m.databaseEnabled {
if err := database.Close(); err != nil {
logging.Warn("failed to close database", logging.Fields{"error": err})
}
}
})
}
// stats, err := manager.GetMinerHistoricalStats("xmrig")
// if err == nil { logging.Info("stats", logging.Fields{"average": stats.AverageRate}) }
func (m *Manager) GetMinerHistoricalStats(minerName string) (*database.HashrateStats, error) {
if !m.databaseEnabled {
return nil, ErrDatabaseError("database persistence is disabled")
}
return database.GetHashrateStats(minerName)
}
// points, err := manager.GetMinerHistoricalHashrate("xmrig", time.Now().Add(-1*time.Hour), time.Now())
func (m *Manager) GetMinerHistoricalHashrate(minerName string, since, until time.Time) ([]HashratePoint, error) {
if !m.databaseEnabled {
return nil, ErrDatabaseError("database persistence is disabled")
}
dbPoints, err := database.GetHashrateHistory(minerName, database.ResolutionHigh, since, until)
if err != nil {
return nil, err
}
// Convert database points to mining points
points := make([]HashratePoint, len(dbPoints))
for i, databasePoint := range dbPoints {
points[i] = HashratePoint{
Timestamp: databasePoint.Timestamp,
Hashrate: databasePoint.Hashrate,
}
}
return points, nil
}
// allStats, err := manager.GetAllMinerHistoricalStats()
// for _, stats := range allStats { logging.Info("stats", logging.Fields{"miner": stats.MinerName, "average": stats.AverageRate}) }
func (m *Manager) GetAllMinerHistoricalStats() ([]database.HashrateStats, error) {
if !m.databaseEnabled {
return nil, ErrDatabaseError("database persistence is disabled")
}
return database.GetAllMinerStats()
}
// if manager.IsDatabaseEnabled() { /* persist stats */ }
func (m *Manager) IsDatabaseEnabled() bool {
return m.databaseEnabled
}