diff --git a/internal/cluster/leader_fallback.go b/internal/cluster/leader_fallback.go new file mode 100644 index 0000000..d287ef6 --- /dev/null +++ b/internal/cluster/leader_fallback.go @@ -0,0 +1,516 @@ +/* + * 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(), + } +}