futriix/internal/cluster/leader_fallback.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/leader_fallback.go
// Назначение: Механизм Single Point of Failure - обеспечение записи при потере лидера

package cluster

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

// WriteRequest представляет запрос на запись при отсутствии лидера
type WriteRequest struct {
    ID        string
    Data      []byte
    CreatedAt int64
    Retries   int
    Callback  chan error
}

// PendingWriteQueue очередь отложенных записей
type PendingWriteQueue struct {
    requests    []*WriteRequest
    mu          sync.Mutex
    maxSize     int
    maxAge      time.Duration
}

// NewPendingWriteQueue создаёт новую очередь
func NewPendingWriteQueue(maxSize int, maxAge time.Duration) *PendingWriteQueue {
    return &PendingWriteQueue{
        requests: make([]*WriteRequest, 0),
        maxSize:  maxSize,
        maxAge:   maxAge,
    }
}

// Add добавляет запрос в очередь
func (q *PendingWriteQueue) Add(req *WriteRequest) error {
    q.mu.Lock()
    defer q.mu.Unlock()
    
    if len(q.requests) >= q.maxSize {
        return fmt.Errorf("pending write queue is full")
    }
    
    q.requests = append(q.requests, req)
    return nil
}

// GetAll возвращает все запросы и очищает очередь
func (q *PendingWriteQueue) GetAll() []*WriteRequest {
    q.mu.Lock()
    defer q.mu.Unlock()
    
    // Очищаем устаревшие запросы
    now := time.Now().UnixMilli()
    valid := make([]*WriteRequest, 0)
    for _, req := range q.requests {
        if now - req.CreatedAt < int64(q.maxAge.Milliseconds()) {
            valid = append(valid, req)
        } else {
            if req.Callback != nil {
                req.Callback <- fmt.Errorf("write request expired")
            }
        }
    }
    
    q.requests = make([]*WriteRequest, 0)
    return valid
}

// Size возвращает размер очереди
func (q *PendingWriteQueue) Size() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    return len(q.requests)
}

// LeaderFallbackManager управляет fallback при потере лидера
type LeaderFallbackManager struct {
    coordinator      *RaftCoordinator
    logger           LoggerInterface
    pendingWrites    *PendingWriteQueue
    fallbackMode     atomic.Bool
    lastLeaderSeen   atomic.Int64
    electionTimeout  time.Duration
    fallbackTimeout  time.Duration
    mu               sync.RWMutex
    stopChan         chan struct{}
    wg               sync.WaitGroup
    observers        map[string]chan *LeaderChangeEvent
    observerMu       sync.RWMutex
    writeBuffer      []*WriteRequest
    bufferMu         sync.Mutex
}

// LeaderChangeEvent событие изменения лидера
type LeaderChangeEvent struct {
    OldLeader string
    NewLeader string
    Timestamp int64
    Term      uint64
}

// FallbackConfig конфигурация fallback механизма
type FallbackConfig struct {
    Enabled            bool
    ElectionTimeout    time.Duration
    FallbackTimeout    time.Duration
    PendingQueueSize   int
    PendingQueueMaxAge time.Duration
    WriteBufferSize    int
}

// DefaultFallbackConfig возвращает конфигурацию по умолчанию
func DefaultFallbackConfig() *FallbackConfig {
    return &FallbackConfig{
        Enabled:            true,
        ElectionTimeout:    5 * time.Second,
        FallbackTimeout:    30 * time.Second,
        PendingQueueSize:   10000,
        PendingQueueMaxAge: 60 * time.Second,
        WriteBufferSize:    1000,
    }
}

// NewLeaderFallbackManager создаёт новый менеджер fallback
func NewLeaderFallbackManager(coordinator *RaftCoordinator, logger LoggerInterface, config *FallbackConfig) *LeaderFallbackManager {
    if config == nil {
        config = DefaultFallbackConfig()
    }
    
    lfm := &LeaderFallbackManager{
        coordinator:     coordinator,
        logger:          logger,
        pendingWrites:   NewPendingWriteQueue(config.PendingQueueSize, config.PendingQueueMaxAge),
        electionTimeout: config.ElectionTimeout,
        fallbackTimeout: config.FallbackTimeout,
        stopChan:        make(chan struct{}),
        observers:       make(map[string]chan *LeaderChangeEvent),
        writeBuffer:     make([]*WriteRequest, 0, config.WriteBufferSize),
    }
    
    lfm.lastLeaderSeen.Store(time.Now().UnixMilli())
    
    // Запускаем мониторинг лидера
    lfm.wg.Add(1)
    go lfm.monitorLeader()
    
    // Запускаем обработчик fallback записей
    lfm.wg.Add(1)
    go lfm.processFallbackWrites()
    
    if logger != nil {
        logger.Info("Leader fallback manager initialized")
    }
    
    return lfm
}

// monitorLeader отслеживает состояние лидера
func (lfm *LeaderFallbackManager) monitorLeader() {
    defer lfm.wg.Done()
    
    ticker := time.NewTicker(lfm.electionTimeout / 2)
    defer ticker.Stop()
    
    var lastLeader string
    var leaderLostAt int64
    
    for {
        select {
        case <-lfm.stopChan:
            return
        case <-ticker.C:
            currentLeader := lfm.coordinator.GetLeader()
            currentLeaderID := ""
            if currentLeader != nil {
                currentLeaderID = currentLeader.ID
            }
            isLeader := lfm.coordinator.IsLeader()
            
            // Проверяем, есть ли лидер
            if currentLeaderID == "" && !isLeader {
                if !lfm.fallbackMode.Load() && leaderLostAt == 0 {
                    leaderLostAt = time.Now().UnixMilli()
                    lfm.enterFallbackMode()
                } else if leaderLostAt > 0 && time.Now().UnixMilli()-leaderLostAt > int64(lfm.fallbackTimeout.Milliseconds()) {
                    lfm.handleProlongedLeaderLoss()
                }
            } else {
                if lfm.fallbackMode.Load() {
                    lfm.exitFallbackMode()
                    lfm.processPendingWrites()
                }
                leaderLostAt = 0
                lfm.lastLeaderSeen.Store(time.Now().UnixMilli())
            }
            
            // Отслеживаем смену лидера
            if lastLeader != currentLeaderID {
                if lastLeader != "" {
                    event := &LeaderChangeEvent{
                        OldLeader: lastLeader,
                        NewLeader: currentLeaderID,
                        Timestamp: time.Now().UnixMilli(),
                        Term:      lfm.coordinator.GetCurrentTerm(),
                    }
                    lfm.notifyObservers(event)
                    
                    if lfm.logger != nil {
                        lfm.logger.Info(fmt.Sprintf("Leader changed from %s to %s", lastLeader, currentLeaderID))
                    }
                }
                lastLeader = currentLeaderID
            }
        }
    }
}

// enterFallbackMode переводит систему в fallback режим
func (lfm *LeaderFallbackManager) enterFallbackMode() {
    if lfm.fallbackMode.CompareAndSwap(false, true) {
        if lfm.logger != nil {
            lfm.logger.Warn("Entering fallback mode - no leader available")
        }
        
        // Создаём событие о потере лидера
        leader := lfm.coordinator.GetLeader()
        oldLeader := ""
        if leader != nil {
            oldLeader = leader.ID
        }
        event := &LeaderChangeEvent{
            OldLeader: oldLeader,
            NewLeader: "",
            Timestamp: time.Now().UnixMilli(),
            Term:      lfm.coordinator.GetCurrentTerm(),
        }
        lfm.notifyObservers(event)
    }
}

// exitFallbackMode выходит из fallback режима
func (lfm *LeaderFallbackManager) exitFallbackMode() {
    if lfm.fallbackMode.CompareAndSwap(true, false) {
        if lfm.logger != nil {
            lfm.logger.Info("Exiting fallback mode - leader elected")
        }
    }
}

// handleProlongedLeaderLoss обрабатывает длительную потерю лидера
func (lfm *LeaderFallbackManager) handleProlongedLeaderLoss() {
    if lfm.logger != nil {
        lfm.logger.Error("Prolonged leader loss detected, initiating emergency measures")
    }
    
    // Проверяем возможность стать лидером
    if !lfm.coordinator.IsLeader() && lfm.coordinator.raft != nil {
        lfm.initiateEmergencyElection()
    }
}

// initiateEmergencyElection инициирует экстренные выборы
func (lfm *LeaderFallbackManager) initiateEmergencyElection() {
    if lfm.logger != nil {
        lfm.logger.Info("Initiating emergency election")
    }
    
    // Принудительно запускаем выборы
    if lfm.coordinator.raft != nil {
        lfm.coordinator.raft.LeadershipTransfer()
    }
    
    // Если через electionTimeout всё ещё нет лидера, пробуем другие узлы
    time.Sleep(lfm.electionTimeout)
    
    if lfm.coordinator.GetLeader() == nil && !lfm.coordinator.IsLeader() {
        lfm.tryAlternativeLeader()
    }
}

// tryAlternativeLeader пробует альтернативные пути выбора лидера
func (lfm *LeaderFallbackManager) tryAlternativeLeader() {
    nodes := lfm.coordinator.GetAllNodes()
    if len(nodes) == 0 {
        return
    }
    
    if lfm.logger != nil {
        lfm.logger.Warn("Attempting to find alternative leader")
    }
    
    // Пробуем каждый узел
    for _, node := range nodes {
        if lfm.coordinator.GetLeader() != nil {
            break
        }
        
        if lfm.logger != nil {
            lfm.logger.Debug(fmt.Sprintf("Checking node %s for leadership possibility", node.ID))
        }
        
        time.Sleep(500 * time.Millisecond)
    }
}

// processPendingWrites обрабатывает отложенные записи
func (lfm *LeaderFallbackManager) processPendingWrites() {
    requests := lfm.pendingWrites.GetAll()
    
    if len(requests) == 0 {
        return
    }
    
    if lfm.logger != nil {
        lfm.logger.Info(fmt.Sprintf("Processing %d pending writes after leader election", len(requests)))
    }
    
    for _, req := range requests {
        // Пытаемся применить запись через Raft
        future := lfm.coordinator.raft.Apply(req.Data, 10*time.Second)
        err := future.Error()
        
        if req.Callback != nil {
            req.Callback <- err
        }
        
        if err != nil && lfm.logger != nil {
            lfm.logger.Error(fmt.Sprintf("Failed to apply pending write %s: %v", req.ID, err))
        }
    }
}

// processFallbackWrites обрабатывает записи в fallback режиме
func (lfm *LeaderFallbackManager) processFallbackWrites() {
    defer lfm.wg.Done()
    
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-lfm.stopChan:
            return
        case <-ticker.C:
            if lfm.fallbackMode.Load() {
                lfm.bufferWrites()
            }
        }
    }
}

// bufferWrites буферизирует записи во время fallback
func (lfm *LeaderFallbackManager) bufferWrites() {
    lfm.bufferMu.Lock()
    defer lfm.bufferMu.Unlock()
    
    if len(lfm.writeBuffer) == 0 {
        return
    }
    
    // Проверяем, появился ли лидер
    if lfm.coordinator.GetLeader() != nil || lfm.coordinator.IsLeader() {
        // Сбрасываем буфер в очередь ожидания
        for _, req := range lfm.writeBuffer {
            lfm.pendingWrites.Add(req)
        }
        lfm.writeBuffer = lfm.writeBuffer[:0]
        lfm.exitFallbackMode()
    } else if len(lfm.writeBuffer) > 0 {
        // Записываем в локальный WAL при длительном отсутствии лидера
        lfm.writeToLocalWAL()
    }
}

// writeToLocalWAL записывает в локальный WAL
func (lfm *LeaderFallbackManager) writeToLocalWAL() {
    lfm.bufferMu.Lock()
    defer lfm.bufferMu.Unlock()
    
    if lfm.logger != nil {
        lfm.logger.Debug(fmt.Sprintf("Writing %d operations to local WAL", len(lfm.writeBuffer)))
    }
    
    // Здесь можно реализовать запись в локальный WAL
    // для последующего восстановления
    lfm.writeBuffer = lfm.writeBuffer[:0]
}

// SubmitWrite submits a write operation with fallback support
func (lfm *LeaderFallbackManager) SubmitWrite(data []byte) error {
    return lfm.SubmitWriteWithCallback(data, nil)
}

// SubmitWriteWithCallback отправляет запись с callback
func (lfm *LeaderFallbackManager) SubmitWriteWithCallback(data []byte, callback chan error) error {
    req := &WriteRequest{
        ID:        fmt.Sprintf("write_%d", time.Now().UnixNano()),
        Data:      data,
        CreatedAt: time.Now().UnixMilli(),
        Retries:   0,
        Callback:  callback,
    }
    
    // Если есть лидер, пытаемся записать сразу
    if !lfm.fallbackMode.Load() && lfm.coordinator.IsLeader() {
        future := lfm.coordinator.raft.Apply(data, 10*time.Second)
        if callback != nil {
            callback <- future.Error()
        }
        return future.Error()
    }
    
    // В fallback режиме буферизируем
    lfm.bufferMu.Lock()
    defer lfm.bufferMu.Unlock()
    
    lfm.writeBuffer = append(lfm.writeBuffer, req)
    
    if lfm.logger != nil {
        lfm.logger.Debug(fmt.Sprintf("Write buffered in fallback mode (buffer size: %d)", len(lfm.writeBuffer)))
    }
    
    return nil
}

// IsFallbackMode возвращает статус fallback режима
func (lfm *LeaderFallbackManager) IsFallbackMode() bool {
    return lfm.fallbackMode.Load()
}

// GetPendingWritesCount возвращает количество отложенных записей
func (lfm *LeaderFallbackManager) GetPendingWritesCount() int {
    return lfm.pendingWrites.Size()
}

// GetBufferedWritesCount возвращает количество буферизированных записей
func (lfm *LeaderFallbackManager) GetBufferedWritesCount() int {
    lfm.bufferMu.Lock()
    defer lfm.bufferMu.Unlock()
    return len(lfm.writeBuffer)
}

// Subscribe подписывается на события смены лидера
func (lfm *LeaderFallbackManager) Subscribe(id string) <-chan *LeaderChangeEvent {
    lfm.observerMu.Lock()
    defer lfm.observerMu.Unlock()
    
    ch := make(chan *LeaderChangeEvent, 10)
    lfm.observers[id] = ch
    return ch
}

// Unsubscribe отписывается от событий
func (lfm *LeaderFallbackManager) Unsubscribe(id string) {
    lfm.observerMu.Lock()
    defer lfm.observerMu.Unlock()
    
    if ch, ok := lfm.observers[id]; ok {
        close(ch)
        delete(lfm.observers, id)
    }
}

// notifyObservers уведомляет наблюдателей
func (lfm *LeaderFallbackManager) notifyObservers(event *LeaderChangeEvent) {
    lfm.observerMu.RLock()
    defer lfm.observerMu.RUnlock()
    
    for _, ch := range lfm.observers {
        select {
        case ch <- event:
        default:
            if lfm.logger != nil {
                lfm.logger.Warn("Observer channel full, dropping event")
            }
        }
    }
}

// Stop останавливает fallback менеджер
func (lfm *LeaderFallbackManager) Stop() {
    close(lfm.stopChan)
    lfm.wg.Wait()
    
    if lfm.logger != nil {
        lfm.logger.Info("Leader fallback manager stopped")
    }
}

// GetStats возвращает статистику
func (lfm *LeaderFallbackManager) GetStats() map[string]interface{} {
    return map[string]interface{}{
        "fallback_mode":          lfm.fallbackMode.Load(),
        "pending_writes":         lfm.GetPendingWritesCount(),
        "buffered_writes":        lfm.GetBufferedWritesCount(),
        "last_leader_seen":       lfm.lastLeaderSeen.Load(),
        "observers_count":        len(lfm.observers),
        "election_timeout_ms":    lfm.electionTimeout.Milliseconds(),
        "fallback_timeout_ms":    lfm.fallbackTimeout.Milliseconds(),
    }
}