futriix/internal/cluster/recovery.go

/*
 * Copyright 2026 Safronov Grigorii
 *
 * Licensed under the CDDL, Version 1.0 (the "License");
 * you may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 * https://opensource.org/licenses/CDDL-1.0
 */

// Файл: internal/cluster/recovery.go
// Назначение: Автоматическое восстановление узлов после сбоя

package cluster

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

// NodeState представляет состояние узла для восстановления
type NodeState struct {
    NodeID        string    `json:"node_id"`
    LastSeen      time.Time `json:"last_seen"`
    LastLogIndex  uint64    `json:"last_log_index"`
    FailureCount  int       `json:"failure_count"`
    IsRecovering  bool      `json:"is_recovering"`
}

// RecoveryManager управляет восстановлением узлов
type RecoveryManager struct {
    coordinator   *RaftCoordinator
    logger        LoggerInterface
    states        sync.Map
    maxFailures   int
    recoveryDelay time.Duration
    stopChan      chan struct{}
    wg            sync.WaitGroup
    isActive      atomic.Bool
}

// NewRecoveryManager создаёт новый менеджер восстановления
func NewRecoveryManager(coordinator *RaftCoordinator, logger LoggerInterface) *RecoveryManager {
    rm := &RecoveryManager{
        coordinator:   coordinator,
        logger:        logger,
        maxFailures:   3,
        recoveryDelay: 30 * time.Second,
        stopChan:      make(chan struct{}),
    }
    
    return rm
}

// Start запускает мониторинг восстановления
func (rm *RecoveryManager) Start() {
    rm.isActive.Store(true)
    rm.wg.Add(2)
    go rm.monitorLoop()
    go rm.recoveryLoop()
    
    if rm.logger != nil {
        rm.logger.Info("Recovery manager started")
    }
}

// Stop останавливает менеджер восстановления
func (rm *RecoveryManager) Stop() {
    rm.isActive.Store(false)
    close(rm.stopChan)
    rm.wg.Wait()
    
    if rm.logger != nil {
        rm.logger.Info("Recovery manager stopped")
    }
}

// monitorLoop отслеживает состояние узлов
func (rm *RecoveryManager) monitorLoop() {
    defer rm.wg.Done()
    
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-rm.stopChan:
            return
        case <-ticker.C:
            rm.checkNodesHealth()
        }
    }
}

// checkNodesHealth проверяет здоровье узлов
func (rm *RecoveryManager) checkNodesHealth() {
    if rm.coordinator == nil {
        return
    }
    
    nodes := rm.coordinator.GetAllNodes()
    now := time.Now()
    
    for _, node := range nodes {
        stateVal, ok := rm.states.Load(node.ID)
        var state *NodeState
        if ok {
            state = stateVal.(*NodeState)
        } else {
            state = &NodeState{
                NodeID:       node.ID,
                LastSeen:     now,
                LastLogIndex: 0,
                FailureCount: 0,
                IsRecovering: false,
            }
            rm.states.Store(node.ID, state)
        }
        
        // Проверяем, был ли недавно heartbeat
        lastSeen := time.UnixMilli(node.LastSeen)
        if now.Sub(lastSeen) > 30*time.Second {
            state.FailureCount++
            if rm.logger != nil {
                rm.logger.Warn(fmt.Sprintf("Node %s appears unhealthy, failure count: %d", node.ID, state.FailureCount))
            }
            
            if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
                rm.triggerRecovery(node.ID)
            }
        } else {
            if state.FailureCount > 0 {
                state.FailureCount = 0
                if rm.logger != nil {
                    rm.logger.Info(fmt.Sprintf("Node %s recovered", node.ID))
                }
            }
        }
        
        state.LastSeen = now
        rm.states.Store(node.ID, state)
    }
}

// triggerRecovery запускает восстановление узла
func (rm *RecoveryManager) triggerRecovery(nodeID string) {
    stateVal, ok := rm.states.Load(nodeID)
    if !ok {
        return
    }
    
    state := stateVal.(*NodeState)
    if state.IsRecovering {
        return
    }
    
    state.IsRecovering = true
    rm.states.Store(nodeID, state)
    
    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Triggering recovery for node %s", nodeID))
    }
    
    // Запускаем асинхронное восстановление
    go rm.recoverNode(nodeID)
}

// recoverNode восстанавливает узел
func (rm *RecoveryManager) recoverNode(nodeID string) {
    defer func() {
        if r := recover(); r != nil {
            if rm.logger != nil {
                rm.logger.Error(fmt.Sprintf("Recovery for node %s panicked: %v", nodeID, r))
            }
        }
        
        // Сбрасываем флаг восстановления
        if stateVal, ok := rm.states.Load(nodeID); ok {
            state := stateVal.(*NodeState)
            state.IsRecovering = false
            rm.states.Store(nodeID, state)
        }
    }()
    
    // Ждём перед попыткой восстановления
    time.Sleep(rm.recoveryDelay)
    
    if rm.coordinator == nil {
        return
    }
    
    // Проверяем, не восстановился ли узел самостоятельно
    node := rm.coordinator.GetNodeByID(nodeID)
    if node != nil && time.Now().UnixMilli()-node.LastSeen < 30000 {
        if rm.logger != nil {
            rm.logger.Info(fmt.Sprintf("Node %s recovered on its own", nodeID))
        }
        return
    }
    
    // Пытаемся переподключить узел
    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Attempting to reconnect node %s", nodeID))
    }
    
    // Обновляем статус узла
    if err := rm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil {
        if rm.logger != nil {
            rm.logger.Error(fmt.Sprintf("Failed to update node %s status: %v", nodeID, err))
        }
    }
    
    // Запускаем синхронизацию данных
    if err := rm.syncNodeData(nodeID); err != nil {
        if rm.logger != nil {
            rm.logger.Error(fmt.Sprintf("Failed to sync node %s data: %v", nodeID, err))
        }
    }
    
    if rm.logger != nil {
        rm.logger.Info(fmt.Sprintf("Recovery completed for node %s", nodeID))
    }
}

// syncNodeData синхронизирует данные с узлом
func (rm *RecoveryManager) syncNodeData(nodeID string) error {
    // Получаем информацию об узле
    node := rm.coordinator.GetNodeByID(nodeID)
    if node == nil {
        return fmt.Errorf("node not found: %s", nodeID)
    }
    
    if rm.logger != nil {
        rm.logger.Debug(fmt.Sprintf("Syncing data with node %s at %s:%d", nodeID, node.IP, node.Port))
    }
    
    // В реальной реализации здесь была бы синхронизация данных
    // Например, отправка всех изменений с момента последнего успешного heartbeat
    
    return nil
}

// recoveryLoop периодически проверяет и восстанавливает узлы
func (rm *RecoveryManager) recoveryLoop() {
    defer rm.wg.Done()
    
    ticker := time.NewTicker(1 * time.Minute)
    defer ticker.Stop()
    
    for {
        select {
        case <-rm.stopChan:
            return
        case <-ticker.C:
            rm.attemptRecoveryAll()
        }
    }
}

// attemptRecoveryAll пытается восстановить все проблемные узлы
func (rm *RecoveryManager) attemptRecoveryAll() {
    rm.states.Range(func(key, value interface{}) bool {
        state := value.(*NodeState)
        if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
            go rm.recoverNode(state.NodeID)
        }
        return true
    })
}

// GetNodeState возвращает состояние узла
func (rm *RecoveryManager) GetNodeState(nodeID string) *NodeState {
    if val, ok := rm.states.Load(nodeID); ok {
        return val.(*NodeState)
    }
    return nil
}

// GetAllStates возвращает все состояния узлов
func (rm *RecoveryManager) GetAllStates() map[string]*NodeState {
    states := make(map[string]*NodeState)
    rm.states.Range(func(key, value interface{}) bool {
        states[key.(string)] = value.(*NodeState)
        return true
    })
    return states
}