From 5d0df1fcf228293186b80dd92568cb7ed5a86f5b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=93=D1=80=D0=B8=D0=B3=D0=BE=D1=80=D0=B8=D0=B9=20=D0=A1?= =?UTF-8?q?=D0=B0=D1=84=D1=80=D0=BE=D0=BD=D0=BE=D0=B2?= Date: Thu, 11 Jun 2026 17:38:45 +0000 Subject: [PATCH] Upload files to "internal/cluster" --- internal/cluster/panic_recovery.go | 364 +++++++++++++++++++++++++++++ 1 file changed, 364 insertions(+) create mode 100644 internal/cluster/panic_recovery.go diff --git a/internal/cluster/panic_recovery.go b/internal/cluster/panic_recovery.go new file mode 100644 index 0000000..4a84612 --- /dev/null +++ b/internal/cluster/panic_recovery.go @@ -0,0 +1,364 @@ +/* + * 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/panic_recovery.go +// Назначение: Автоматическое восстановление после паники по всему коду + +package cluster + +import ( + "fmt" + "runtime" + "runtime/debug" + "sync" + "sync/atomic" + "time" +) + +// PanicInfo содержит информацию о панике +type PanicInfo struct { + ID string + GoroutineID int64 + PanicValue interface{} + StackTrace string + Timestamp int64 + Recovered bool + RecoveryTime int64 +} + +// PanicRecoveryManager управляет восстановлением после паник +type PanicRecoveryManager struct { + mu sync.RWMutex + panics map[string]*PanicInfo + maxPanics int + recoveryFuncs map[string]func(interface{}) error + logger LoggerInterface + stopChan chan struct{} + wg sync.WaitGroup + totalPanics atomic.Uint64 + recoveredFrom atomic.Uint64 + failedRecoveries atomic.Uint64 +} + +// NewPanicRecoveryManager создаёт новый менеджер восстановления +func NewPanicRecoveryManager(logger LoggerInterface) *PanicRecoveryManager { + prm := &PanicRecoveryManager{ + panics: make(map[string]*PanicInfo), + maxPanics: 1000, + recoveryFuncs: make(map[string]func(interface{}) error), + logger: logger, + stopChan: make(chan struct{}), + } + + prm.registerDefaultRecoveryFuncs() + + // Запускаем периодическую очистку старых паник + prm.wg.Add(1) + go prm.cleanupOldPanics() + + if logger != nil { + logger.Info("Panic recovery manager initialized") + } + + return prm +} + +// getGoroutineID возвращает ID текущей горутины +func getGoroutineID() int64 { + var buf [64]byte + n := runtime.Stack(buf[:], false) + var id int64 + fmt.Sscanf(string(buf[:n]), "goroutine %d", &id) + return id +} + +// RegisterRecoveryFunc регистрирует функцию восстановления для компонента +func (prm *PanicRecoveryManager) RegisterRecoveryFunc(component string, fn func(interface{}) error) { + prm.mu.Lock() + defer prm.mu.Unlock() + prm.recoveryFuncs[component] = fn +} + +// registerDefaultRecoveryFuncs регистрирует стандартные функции восстановления +func (prm *PanicRecoveryManager) registerDefaultRecoveryFuncs() { + // Восстановление для TCP сервера + prm.recoveryFuncs["TCPServer"] = func(p interface{}) error { + return nil // Базовое восстановление + } + + // Восстановление для обработчика соединений + prm.recoveryFuncs["ConnectionHandler"] = func(p interface{}) error { + return nil + } + + // Восстановление для репликации + prm.recoveryFuncs["Replication"] = func(p interface{}) error { + return nil + } + + // Восстановление для WAL + prm.recoveryFuncs["WAL"] = func(p interface{}) error { + return nil + } + + // Восстановление для Raft + prm.recoveryFuncs["Raft"] = func(p interface{}) error { + return nil + } +} + +// Recover обрабатывает панику и пытается восстановиться +func (prm *PanicRecoveryManager) Recover(component string, context map[string]interface{}) { + if r := recover(); r != nil { + prm.totalPanics.Add(1) + + panicInfo := &PanicInfo{ + ID: fmt.Sprintf("panic_%d_%d", time.Now().UnixNano(), prm.totalPanics.Load()), + GoroutineID: getGoroutineID(), + PanicValue: r, + StackTrace: string(debug.Stack()), + Timestamp: time.Now().UnixMilli(), + Recovered: false, + } + + prm.mu.Lock() + // Ограничиваем количество сохраняемых паник + if len(prm.panics) >= prm.maxPanics { + for k := range prm.panics { + delete(prm.panics, k) + break + } + } + prm.panics[panicInfo.ID] = panicInfo + prm.mu.Unlock() + + if prm.logger != nil { + prm.logger.Error(fmt.Sprintf("PANIC in component %s: %v\n%s", component, r, panicInfo.StackTrace)) + } + + // Пытаемся восстановиться + if recoveryFn, ok := prm.recoveryFuncs[component]; ok { + if err := recoveryFn(r); err != nil { + prm.failedRecoveries.Add(1) + if prm.logger != nil { + prm.logger.Error(fmt.Sprintf("Failed to recover from panic in %s: %v", component, err)) + } + } else { + panicInfo.Recovered = true + panicInfo.RecoveryTime = time.Now().UnixMilli() + prm.recoveredFrom.Add(1) + + if prm.logger != nil { + prm.logger.Info(fmt.Sprintf("Successfully recovered from panic in %s", component)) + } + } + } else { + if prm.logger != nil { + prm.logger.Error(fmt.Sprintf("No recovery function registered for component %s", component)) + } + } + } +} + +// RecoverWithRetry пытается восстановиться с повторными попытками +func (prm *PanicRecoveryManager) RecoverWithRetry(component string, retries int, fn func() error) error { + var lastErr error + + for i := 0; i < retries; i++ { + func() { + defer prm.Recover(component, map[string]interface{}{ + "attempt": i + 1, + "retries": retries, + }) + lastErr = fn() + }() + + if lastErr == nil { + return nil + } + + if i < retries-1 { + time.Sleep(time.Duration(100*(i+1)) * time.Millisecond) + } + } + + return lastErr +} + +// SafeGo безопасно запускает горутину с восстановлением +func (prm *PanicRecoveryManager) SafeGo(component string, fn func()) { + go func() { + defer prm.Recover(component, nil) + fn() + }() +} + +// SafeGoWithContext безопасно запускает горутину с контекстом +func (prm *PanicRecoveryManager) SafeGoWithContext(component string, ctx map[string]interface{}, fn func()) { + go func() { + defer prm.Recover(component, ctx) + fn() + }() +} + +// cleanupOldPanics периодически очищает старые записи о паниках +func (prm *PanicRecoveryManager) cleanupOldPanics() { + defer prm.wg.Done() + + ticker := time.NewTicker(1 * time.Hour) + defer ticker.Stop() + + for { + select { + case <-prm.stopChan: + return + case <-ticker.C: + prm.mu.Lock() + now := time.Now().UnixMilli() + for id, info := range prm.panics { + // Удаляем паники старше 24 часов + if now-info.Timestamp > 24*3600*1000 { + delete(prm.panics, id) + } + } + prm.mu.Unlock() + } + } +} + +// GetPanicInfo возвращает информацию о панике +func (prm *PanicRecoveryManager) GetPanicInfo(id string) *PanicInfo { + prm.mu.RLock() + defer prm.mu.RUnlock() + return prm.panics[id] +} + +// GetAllPanics возвращает все паники +func (prm *PanicRecoveryManager) GetAllPanics() []*PanicInfo { + prm.mu.RLock() + defer prm.mu.RUnlock() + + result := make([]*PanicInfo, 0, len(prm.panics)) + for _, info := range prm.panics { + result = append(result, info) + } + return result +} + +// GetStats возвращает статистику +func (prm *PanicRecoveryManager) GetStats() map[string]interface{} { + return map[string]interface{}{ + "total_panics": prm.totalPanics.Load(), + "recovered_from": prm.recoveredFrom.Load(), + "failed_recoveries": prm.failedRecoveries.Load(), + "active_panics": len(prm.panics), + "max_panics": prm.maxPanics, + } +} + +// Stop останавливает менеджер +func (prm *PanicRecoveryManager) Stop() { + close(prm.stopChan) + prm.wg.Wait() +} + +// RecoverableRoutine обёртка для восстанавливаемых горутин +type RecoverableRoutine struct { + name string + fn func() error + mgr *PanicRecoveryManager + maxRetries int + stopChan chan struct{} + running atomic.Bool + mu sync.Mutex +} + +// NewRecoverableRoutine создаёт новую восстанавливаемую горутину +func NewRecoverableRoutine(name string, fn func() error, mgr *PanicRecoveryManager, maxRetries int) *RecoverableRoutine { + return &RecoverableRoutine{ + name: name, + fn: fn, + mgr: mgr, + maxRetries: maxRetries, + stopChan: make(chan struct{}), + } +} + +// Start запускает горутину с автоматическим восстановлением +func (rr *RecoverableRoutine) Start() { + if !rr.running.CompareAndSwap(false, true) { + return + } + + go rr.run() +} + +// Stop останавливает горутину +func (rr *RecoverableRoutine) Stop() { + close(rr.stopChan) + rr.running.Store(false) +} + +// run запускает основной цикл с восстановлением +func (rr *RecoverableRoutine) run() { + defer rr.running.Store(false) + + retries := 0 + + for { + select { + case <-rr.stopChan: + return + default: + err := rr.runWithRecovery() + if err == nil { + retries = 0 + } else { + retries++ + if retries > rr.maxRetries { + if rr.mgr.logger != nil { + rr.mgr.logger.Error(fmt.Sprintf("Routine %s exceeded max retries (%d)", rr.name, rr.maxRetries)) + } + return + } + + backoff := time.Duration(100*retries) * time.Millisecond + if backoff > 5*time.Second { + backoff = 5 * time.Second + } + + select { + case <-rr.stopChan: + return + case <-time.After(backoff): + } + } + } + } +} + +// runWithRecovery запускает функцию с защитой от паники +func (rr *RecoverableRoutine) runWithRecovery() (err error) { + defer func() { + if r := recover(); r != nil { + rr.mgr.Recover(rr.name, map[string]interface{}{ + "routine": rr.name, + }) + err = fmt.Errorf("panic recovered: %v", r) + } + }() + + return rr.fn() +} + +// IsRunning возвращает статус работы +func (rr *RecoverableRoutine) IsRunning() bool { + return rr.running.Load() +}