From c858b1eec394b3d4b3f4f4e2a86935f60fbf8263 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, 25 Jun 2026 20:49:44 +0000 Subject: [PATCH] Upload files to "internal/backup" --- internal/backup/backup_scheduler.go | 1299 +++++++++++++++++++++++++++ 1 file changed, 1299 insertions(+) create mode 100644 internal/backup/backup_scheduler.go diff --git a/internal/backup/backup_scheduler.go b/internal/backup/backup_scheduler.go new file mode 100644 index 0000000..992b78e --- /dev/null +++ b/internal/backup/backup_scheduler.go @@ -0,0 +1,1299 @@ +/* + * 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/backup/backup_scheduler.go +// Назначение: Планировщик бэкапов и инкрементальные бэкапы на основе WAL +// Использует 2PC для инкрементных бекапов и статистику по 2PC транзакциям + +package backup + +import ( + "bytes" + "compress/gzip" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "sync" + "sync/atomic" + "time" + + "futriis/internal/storage" +) + +// BackupType тип бэкапа +type BackupType string + +const ( + FullBackup BackupType = "full" + IncrementalBackup BackupType = "incremental" +) + +// BackupSchedule расписание бэкапов +type BackupSchedule struct { + ID string `json:"id"` + Name string `json:"name"` + CronExpr string `json:"cron_expr"` + Type BackupType `json:"type"` + RetentionDays int `json:"retention_days"` + Enabled bool `json:"enabled"` + LastRun int64 `json:"last_run"` + NextRun int64 `json:"next_run"` + CreatedAt int64 `json:"created_at"` + UpdatedAt int64 `json:"updated_at"` +} + +// BackupInfo информация о бэкапе +type BackupInfo struct { + ID string `json:"id"` + ScheduleID string `json:"schedule_id"` + Type BackupType `json:"type"` + StartTime int64 `json:"start_time"` + EndTime int64 `json:"end_time"` + Status string `json:"status"` + SizeBytes int64 `json:"size_bytes"` + Path string `json:"path"` + WALStart uint64 `json:"wal_start"` + WALEnd uint64 `json:"wal_end"` + ParentID string `json:"parent_id,omitempty"` + Checksum string `json:"checksum"` + Error string `json:"error,omitempty"` + // ДОБАВЛЕНО: информация о 2PC состоянии + TwoPCCount int `json:"two_pc_count,omitempty"` +} + +// WALEntry запись WAL +type WALEntry struct { + Index uint64 `json:"index"` + Term uint64 `json:"term"` + Type string `json:"type"` + Data []byte `json:"data"` + Timestamp int64 `json:"timestamp"` +} + +// WALReader интерфейс для чтения WAL +type WALReader interface { + ReadSince(index uint64) ([]WALEntry, error) + GetCurrentIndex() (uint64, error) + GetLastBackupIndex() uint64 + SetLastBackupIndex(index uint64) error + GetSegments() ([]string, error) + ReadSegment(segmentPath string) ([]WALEntry, error) +} + +// BackupStorage интерфейс для хранения бэкапов +type BackupStorage interface { + SaveBackup(path string, data []byte) error + LoadBackup(path string) ([]byte, error) + ListBackups() ([]string, error) + DeleteBackup(path string) error + BackupExists(path string) bool +} + +// LoggerInterface интерфейс для логирования +type LoggerInterface interface { + Info(msg string) + Warn(msg string) + Error(msg string) + Debug(msg string) +} + +// FileBackupStorage реализация BackupStorage для файловой системы +type FileBackupStorage struct { + backupDir string + compressLevel int + mu sync.Mutex +} + +// NewFileBackupStorage создаёт новое файловое хранилище бэкапов +func NewFileBackupStorage(backupDir string, compressLevel int) (*FileBackupStorage, error) { + if compressLevel < 1 || compressLevel > 9 { + compressLevel = 6 + } + if err := os.MkdirAll(backupDir, 0755); err != nil { + return nil, fmt.Errorf("failed to create backup dir: %v", err) + } + return &FileBackupStorage{ + backupDir: backupDir, + compressLevel: compressLevel, + }, nil +} + +func (fbs *FileBackupStorage) SaveBackup(path string, data []byte) error { + fbs.mu.Lock() + defer fbs.mu.Unlock() + + fullPath := filepath.Join(fbs.backupDir, path) + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + return err + } + + // Сжимаем данные + var buf bytes.Buffer + gzWriter, err := gzip.NewWriterLevel(&buf, fbs.compressLevel) + if err != nil { + return err + } + if _, err := gzWriter.Write(data); err != nil { + gzWriter.Close() + return err + } + if err := gzWriter.Close(); err != nil { + return err + } + + compressedPath := fullPath + ".gz" + return os.WriteFile(compressedPath, buf.Bytes(), 0644) +} + +func (fbs *FileBackupStorage) LoadBackup(path string) ([]byte, error) { + fbs.mu.Lock() + defer fbs.mu.Unlock() + + fullPath := filepath.Join(fbs.backupDir, path) + compressedPath := fullPath + ".gz" + + var data []byte + var err error + + // Пробуем сначала сжатый файл + if _, err := os.Stat(compressedPath); err == nil { + compressedData, err := os.ReadFile(compressedPath) + if err != nil { + return nil, err + } + gzReader, err := gzip.NewReader(bytes.NewReader(compressedData)) + if err != nil { + return nil, err + } + defer gzReader.Close() + data, err = io.ReadAll(gzReader) + if err != nil { + return nil, err + } + } else { + data, err = os.ReadFile(fullPath) + if err != nil { + return nil, err + } + } + + return data, nil +} + +func (fbs *FileBackupStorage) ListBackups() ([]string, error) { + entries, err := os.ReadDir(fbs.backupDir) + if err != nil { + return nil, err + } + names := make([]string, 0, len(entries)) + for _, e := range entries { + if e.IsDir() { + // Рекурсивно собираем файлы из поддиректорий + subEntries, err := os.ReadDir(filepath.Join(fbs.backupDir, e.Name())) + if err == nil { + for _, sub := range subEntries { + if !sub.IsDir() { + names = append(names, filepath.Join(e.Name(), sub.Name())) + } + } + } + } else { + names = append(names, e.Name()) + } + } + return names, nil +} + +func (fbs *FileBackupStorage) DeleteBackup(path string) error { + fbs.mu.Lock() + defer fbs.mu.Unlock() + + fullPath := filepath.Join(fbs.backupDir, path) + compressedPath := fullPath + ".gz" + + var err error + if _, statErr := os.Stat(compressedPath); statErr == nil { + err = os.Remove(compressedPath) + } else { + err = os.Remove(fullPath) + } + + // Пробуем удалить пустые директории + dirPath := filepath.Dir(fullPath) + os.Remove(dirPath) + + return err +} + +func (fbs *FileBackupStorage) BackupExists(path string) bool { + fullPath := filepath.Join(fbs.backupDir, path) + compressedPath := fullPath + ".gz" + _, err1 := os.Stat(fullPath) + _, err2 := os.Stat(compressedPath) + return err1 == nil || err2 == nil +} + +// WALReaderImpl реализация WALReader для SegmentedWALManager +type WALReaderImpl struct { + walManager interface{} // *SegmentedWALManager или *WALManager + lastBackupIndex atomic.Uint64 + segmentsDir string + mu sync.RWMutex + logger LoggerInterface +} + +// NewWALReaderImpl создаёт новый WALReader +func NewWALReaderImpl(walManager interface{}, segmentsDir string, logger LoggerInterface) *WALReaderImpl { + return &WALReaderImpl{ + walManager: walManager, + segmentsDir: segmentsDir, + logger: logger, + } +} + +// ReadSince читает записи WAL с указанного индекса +func (wr *WALReaderImpl) ReadSince(index uint64) ([]WALEntry, error) { + wr.mu.RLock() + defer wr.mu.RUnlock() + + entries := make([]WALEntry, 0) + + // Получаем список сегментов + segments, err := wr.GetSegments() + if err != nil { + return nil, err + } + + for _, segmentPath := range segments { + segEntries, err := wr.ReadSegment(segmentPath) + if err != nil { + if wr.logger != nil { + wr.logger.Warn(fmt.Sprintf("Failed to read segment %s: %v", segmentPath, err)) + } + continue + } + + for _, entry := range segEntries { + if entry.Index > index { + entries = append(entries, entry) + } + } + } + + return entries, nil +} + +// GetCurrentIndex возвращает текущий индекс WAL +func (wr *WALReaderImpl) GetCurrentIndex() (uint64, error) { + segments, err := wr.GetSegments() + if err != nil { + return 0, err + } + + if len(segments) == 0 { + return 0, nil + } + + // Читаем последний сегмент для получения последнего индекса + lastSegment := segments[len(segments)-1] + entries, err := wr.ReadSegment(lastSegment) + if err != nil { + return 0, err + } + + if len(entries) == 0 { + return 0, nil + } + + return entries[len(entries)-1].Index, nil +} + +// GetLastBackupIndex возвращает последний сохранённый индекс бэкапа +func (wr *WALReaderImpl) GetLastBackupIndex() uint64 { + return wr.lastBackupIndex.Load() +} + +// SetLastBackupIndex устанавливает последний индекс бэкапа +func (wr *WALReaderImpl) SetLastBackupIndex(index uint64) error { + wr.lastBackupIndex.Store(index) + + // Сохраняем в файл для персистентности + indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json") + data, err := json.Marshal(map[string]uint64{"last_index": index}) + if err != nil { + return err + } + return os.WriteFile(indexPath, data, 0644) +} + +// GetSegments возвращает список сегментов WAL +func (wr *WALReaderImpl) GetSegments() ([]string, error) { + if wr.segmentsDir == "" { + return nil, fmt.Errorf("segments directory not set") + } + + pattern := filepath.Join(wr.segmentsDir, "wal_segment_*.log") + files, err := filepath.Glob(pattern) + if err != nil { + return nil, err + } + + sort.Strings(files) + return files, nil +} + +// ReadSegment читает сегмент WAL +func (wr *WALReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { + data, err := os.ReadFile(segmentPath) + if err != nil { + return nil, err + } + + entries := make([]WALEntry, 0) + + // Парсим WAL записи (формат: [length][data] повторяется) + pos := 0 + for pos < len(data) { + if pos+4 > len(data) { + break + } + + // Читаем длину записи (4 байта, big-endian) + length := int(data[pos])<<24 | int(data[pos+1])<<16 | int(data[pos+2])<<8 | int(data[pos+3]) + pos += 4 + + if pos+length > len(data) { + break + } + + recordData := data[pos : pos+length] + pos += length + + var record struct { + LSN uint64 `json:"lsn"` + Timestamp int64 `json:"timestamp"` + Type byte `json:"type"` + Data []byte `json:"data"` + } + + if err := json.Unmarshal(recordData, &record); err != nil { + continue + } + + entry := WALEntry{ + Index: record.LSN, + Timestamp: record.Timestamp, + Type: fmt.Sprintf("%d", record.Type), + Data: record.Data, + } + + entries = append(entries, entry) + } + + return entries, nil +} + +// loadLastBackupIndex загружает последний индекс из файла +func (wr *WALReaderImpl) loadLastBackupIndex() { + indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json") + data, err := os.ReadFile(indexPath) + if err != nil { + return + } + + var meta map[string]uint64 + if err := json.Unmarshal(data, &meta); err != nil { + return + } + + if lastIdx, ok := meta["last_index"]; ok { + wr.lastBackupIndex.Store(lastIdx) + } +} + +// BackupScheduler планировщик бэкапов +type BackupScheduler struct { + schedules map[string]*BackupSchedule + backups map[string]*BackupInfo + walReader WALReader + storage BackupStorage + logger LoggerInterface + mu sync.RWMutex + stopChan chan struct{} + wg sync.WaitGroup + running atomic.Bool + backupCounter atomic.Uint64 + config *BackupSchedulerConfig + // ДОБАВЛЕНО: ссылка на менеджер транзакций для 2PC + txManager *storage.TransactionManager +} + +// BackupSchedulerConfig конфигурация планировщика +type BackupSchedulerConfig struct { + BackupDir string `json:"backup_dir"` + MaxConcurrent int `json:"max_concurrent"` + CompressEnabled bool `json:"compress_enabled"` + CompressionLevel int `json:"compression_level"` + MaxBackupSizeMB int64 `json:"max_backup_size_mb"` + EnableIncremental bool `json:"enable_incremental"` + // ДОБАВЛЕНО: включить сохранение 2PC состояния + IncludeTwoPCState bool `json:"include_two_pc_state"` +} + +// DefaultBackupSchedulerConfig возвращает конфигурацию по умолчанию +func DefaultBackupSchedulerConfig() *BackupSchedulerConfig { + return &BackupSchedulerConfig{ + BackupDir: "backups", + MaxConcurrent: 1, + CompressEnabled: true, + CompressionLevel: 6, + MaxBackupSizeMB: 10240, // 10GB + EnableIncremental: true, + IncludeTwoPCState: true, + } +} + +// NewBackupScheduler создаёт новый планировщик бэкапов +func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger LoggerInterface) (*BackupScheduler, error) { + if cfg == nil { + cfg = DefaultBackupSchedulerConfig() + } + + compressLevel := cfg.CompressionLevel + if !cfg.CompressEnabled { + compressLevel = 0 + } + + storage, err := NewFileBackupStorage(cfg.BackupDir, compressLevel) + if err != nil { + return nil, err + } + + bs := &BackupScheduler{ + schedules: make(map[string]*BackupSchedule), + backups: make(map[string]*BackupInfo), + walReader: walReader, + storage: storage, + logger: logger, + stopChan: make(chan struct{}), + config: cfg, + // ДОБАВЛЕНО: получаем менеджер транзакций + txManager: storage.GetTransactionManager(), + } + + // Загружаем существующие расписания + bs.loadSchedules() + + // Загружаем историю бэкапов + bs.loadBackupHistory() + + return bs, nil +} + +// GetTransactionManager возвращает менеджер транзакций (ДОБАВЛЕНО) +func GetTransactionManager() *storage.TransactionManager { + return storage.GetTransactionManager() +} + +// loadSchedules загружает расписания из хранилища +func (bs *BackupScheduler) loadSchedules() { + data, err := bs.storage.LoadBackup("schedules.json") + if err != nil { + if !os.IsNotExist(err) && bs.logger != nil { + bs.logger.Warn(fmt.Sprintf("Failed to load schedules: %v", err)) + } + return + } + + var schedules map[string]*BackupSchedule + if err := json.Unmarshal(data, &schedules); err != nil { + if bs.logger != nil { + bs.logger.Warn(fmt.Sprintf("Failed to parse schedules: %v", err)) + } + return + } + + bs.mu.Lock() + bs.schedules = schedules + bs.mu.Unlock() +} + +// saveSchedules сохраняет расписания +func (bs *BackupScheduler) saveSchedules() error { + bs.mu.RLock() + data, err := json.MarshalIndent(bs.schedules, "", " ") + bs.mu.RUnlock() + + if err != nil { + return err + } + + return bs.storage.SaveBackup("schedules.json", data) +} + +// loadBackupHistory загружает историю бэкапов +func (bs *BackupScheduler) loadBackupHistory() { + data, err := bs.storage.LoadBackup("backups.json") + if err != nil { + if !os.IsNotExist(err) && bs.logger != nil { + bs.logger.Warn(fmt.Sprintf("Failed to load backup history: %v", err)) + } + return + } + + var backups map[string]*BackupInfo + if err := json.Unmarshal(data, &backups); err != nil { + if bs.logger != nil { + bs.logger.Warn(fmt.Sprintf("Failed to parse backup history: %v", err)) + } + return + } + + bs.mu.Lock() + bs.backups = backups + bs.mu.Unlock() +} + +// saveBackupHistory сохраняет историю бэкапов +func (bs *BackupScheduler) saveBackupHistory() error { + bs.mu.RLock() + data, err := json.MarshalIndent(bs.backups, "", " ") + bs.mu.RUnlock() + + if err != nil { + return err + } + + return bs.storage.SaveBackup("backups.json", data) +} + +// AddSchedule добавляет расписание бэкапов +func (bs *BackupScheduler) AddSchedule(name, cronExpr string, backupType BackupType, retentionDays int) (*BackupSchedule, error) { + bs.mu.Lock() + defer bs.mu.Unlock() + + // Проверяем, что расписание с таким именем не существует + for _, s := range bs.schedules { + if s.Name == name { + return nil, fmt.Errorf("schedule with name '%s' already exists", name) + } + } + + schedule := &BackupSchedule{ + ID: fmt.Sprintf("sch_%d", time.Now().UnixNano()), + Name: name, + CronExpr: cronExpr, + Type: backupType, + RetentionDays: retentionDays, + Enabled: true, + CreatedAt: time.Now().UnixMilli(), + UpdatedAt: time.Now().UnixMilli(), + } + + bs.schedules[schedule.ID] = schedule + + if err := bs.saveSchedules(); err != nil { + return nil, fmt.Errorf("failed to save schedules: %v", err) + } + + if bs.logger != nil { + bs.logger.Info(fmt.Sprintf("Added backup schedule: %s (%s) with cron %s", name, backupType, cronExpr)) + } + return schedule, nil +} + +// RemoveSchedule удаляет расписание +func (bs *BackupScheduler) RemoveSchedule(scheduleID string) error { + bs.mu.Lock() + defer bs.mu.Unlock() + + if _, exists := bs.schedules[scheduleID]; !exists { + return fmt.Errorf("schedule not found: %s", scheduleID) + } + + delete(bs.schedules, scheduleID) + + if err := bs.saveSchedules(); err != nil { + return fmt.Errorf("failed to save schedules: %v", err) + } + + if bs.logger != nil { + bs.logger.Info(fmt.Sprintf("Removed schedule: %s", scheduleID)) + } + return nil +} + +// RunNow запускает бэкап немедленно +func (bs *BackupScheduler) RunNow(scheduleID string) error { + bs.mu.RLock() + schedule, exists := bs.schedules[scheduleID] + bs.mu.RUnlock() + + if !exists { + return fmt.Errorf("schedule not found: %s", scheduleID) + } + + if !bs.config.EnableIncremental && schedule.Type == IncrementalBackup { + return fmt.Errorf("incremental backups are disabled in configuration") + } + + if schedule.Type == FullBackup { + return bs.performFullBackup(scheduleID) + } + return bs.performIncrementalBackup(scheduleID) +} + +// performFullBackup выполняет полный бэкап +func (bs *BackupScheduler) performFullBackup(scheduleID string) error { + if !bs.running.CompareAndSwap(false, true) { + return fmt.Errorf("backup already in progress") + } + defer bs.running.Store(false) + + startTime := time.Now().UnixMilli() + backupID := fmt.Sprintf("full_%d_%d", startTime, bs.backupCounter.Add(1)) + + bs.logger.Info(fmt.Sprintf("Starting full backup %s", backupID)) + + // ========== ДОБАВЛЕНО: Блокировка для 2PC ========== + if bs.txManager != nil && bs.config.IncludeTwoPCState { + bs.logger.Info("Locking transaction manager for backup...") + bs.txManager.LockForBackup() + defer bs.txManager.UnlockForBackup() + } + + // Получаем текущую позицию WAL + walIndex, err := bs.walReader.GetCurrentIndex() + if err != nil { + return fmt.Errorf("failed to get WAL index: %v", err) + } + + // ========== ДОБАВЛЕНО: Получаем незавершённые 2PC транзакции ========== + var pending2PC []*storage.TransactionRecord + var twoPCCount int + if bs.txManager != nil && bs.config.IncludeTwoPCState { + pending2PC = bs.txManager.GetPending2PCTransactions() + twoPCCount = len(pending2PC) + if twoPCCount > 0 { + bs.logger.Info(fmt.Sprintf("Found %d pending 2PC transactions, including in backup", twoPCCount)) + } + } + + // Сохраняем последний индекс для инкрементальных бэкапов + if err := bs.walReader.SetLastBackupIndex(walIndex); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err)) + } + + // Читаем все WAL записи для полного бэкапа + walEntries, err := bs.walReader.ReadSince(0) + if err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to read WAL for full backup: %v", err)) + walEntries = []WALEntry{} + } + + // ========== ДОБАВЛЕНО: Сохраняем состояние 2PC ========== + twoPCState := make(map[string]interface{}) + if twoPCCount > 0 && bs.config.IncludeTwoPCState { + // Сохраняем только метаданные, не сами транзакции (они уже в WAL) + twoPCState["pending_count"] = twoPCCount + twoPCState["timestamp"] = time.Now().UnixMilli() + twoPCState["transaction_ids"] = make([]string, 0, twoPCCount) + for _, tx := range pending2PC { + twoPCState["transaction_ids"] = append(twoPCState["transaction_ids"].([]string), fmt.Sprintf("%d", tx.ID)) + } + } + + // Создаём структуру бэкапа + backupData := struct { + BackupID string `json:"backup_id"` + Type string `json:"type"` + StartTime int64 `json:"start_time"` + WALIndex uint64 `json:"wal_index"` + Entries []WALEntry `json:"entries,omitempty"` + Metadata interface{} `json:"metadata,omitempty"` + TwoPCState interface{} `json:"two_pc_state,omitempty"` + }{ + BackupID: backupID, + Type: "full", + StartTime: startTime, + WALIndex: walIndex, + Entries: walEntries, + Metadata: map[string]interface{}{ + "version": "1.0", + "created_by": "backup_scheduler", + "wal_entries_count": len(walEntries), + "two_pc_pending_count": twoPCCount, + "include_two_pc": bs.config.IncludeTwoPCState, + }, + TwoPCState: twoPCState, + } + + data, err := json.Marshal(backupData) + if err != nil { + return fmt.Errorf("failed to marshal backup data: %v", err) + } + + // Проверяем размер бэкапа + if bs.config.MaxBackupSizeMB > 0 && int64(len(data)) > bs.config.MaxBackupSizeMB*1024*1024 { + return fmt.Errorf("backup size %d exceeds maximum allowed %d MB", + len(data)/(1024*1024), bs.config.MaxBackupSizeMB) + } + + // Сохраняем бэкап + backupPath := fmt.Sprintf("full/%s/%s.backup", + time.Now().Format("2006-01-02"), backupID) + + if err := bs.storage.SaveBackup(backupPath, data); err != nil { + return fmt.Errorf("failed to save backup: %v", err) + } + + // Вычисляем контрольную сумму + hash := sha256.Sum256(data) + checksum := hex.EncodeToString(hash[:]) + + endTime := time.Now().UnixMilli() + + backupInfo := &BackupInfo{ + ID: backupID, + ScheduleID: scheduleID, + Type: FullBackup, + StartTime: startTime, + EndTime: endTime, + Status: "completed", + SizeBytes: int64(len(data)), + Path: backupPath, + WALStart: 0, + WALEnd: walIndex, + Checksum: checksum, + TwoPCCount: twoPCCount, + } + + bs.mu.Lock() + bs.backups[backupID] = backupInfo + bs.mu.Unlock() + + if err := bs.saveBackupHistory(); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err)) + } + + bs.updateScheduleLastRun(scheduleID, startTime) + bs.cleanupOldBackups(scheduleID) + + duration := (endTime - startTime) / 1000 + bs.logger.Info(fmt.Sprintf("Full backup %s completed in %d seconds, size=%d bytes, entries=%d, pending_2pc=%d", + backupID, duration, backupInfo.SizeBytes, len(walEntries), twoPCCount)) + + return nil +} + +// performIncrementalBackup выполняет инкрементальный бэкап на основе WAL +func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error { + if !bs.running.CompareAndSwap(false, true) { + return fmt.Errorf("backup already in progress") + } + defer bs.running.Store(false) + + startTime := time.Now().UnixMilli() + backupID := fmt.Sprintf("inc_%d_%d", startTime, bs.backupCounter.Add(1)) + + // ========== ДОБАВЛЕНО: Блокировка для 2PC ========== + if bs.txManager != nil && bs.config.IncludeTwoPCState { + bs.logger.Info("Locking transaction manager for incremental backup...") + bs.txManager.LockForBackup() + defer bs.txManager.UnlockForBackup() + } + + // Находим последний полный бэкап + lastFullBackup := bs.findLastFullBackup() + if lastFullBackup == nil { + bs.logger.Info("No full backup found, performing full backup instead") + return bs.performFullBackup(scheduleID) + } + + // Проверяем, что родительский бэкап существует + if !bs.storage.BackupExists(lastFullBackup.Path) { + bs.logger.Warn(fmt.Sprintf("Parent backup %s file missing, performing full backup instead", lastFullBackup.ID)) + return bs.performFullBackup(scheduleID) + } + + // Получаем WAL записи с момента последнего бэкапа + lastIndex := bs.walReader.GetLastBackupIndex() + if lastIndex == 0 { + lastIndex = lastFullBackup.WALEnd + } + + walEntries, err := bs.walReader.ReadSince(lastIndex) + if err != nil { + return fmt.Errorf("failed to read WAL: %v", err) + } + + if len(walEntries) == 0 { + bs.logger.Info("No new WAL entries since last backup") + return nil + } + + currentIndex, err := bs.walReader.GetCurrentIndex() + if err != nil { + return fmt.Errorf("failed to get current WAL index: %v", err) + } + + // ========== ДОБАВЛЕНО: Получаем незавершённые 2PC транзакции ========== + var pending2PC []*storage.TransactionRecord + var twoPCCount int + if bs.txManager != nil && bs.config.IncludeTwoPCState { + pending2PC = bs.txManager.GetPending2PCTransactions() + twoPCCount = len(pending2PC) + if twoPCCount > 0 { + bs.logger.Info(fmt.Sprintf("Found %d pending 2PC transactions in incremental backup", twoPCCount)) + } + } + + // Создаём структуру инкрементального бэкапа + backupData := struct { + BackupID string `json:"backup_id"` + Type string `json:"type"` + StartTime int64 `json:"start_time"` + WALStart uint64 `json:"wal_start"` + WALEnd uint64 `json:"wal_end"` + ParentID string `json:"parent_id"` + Entries []WALEntry `json:"entries"` + Metadata interface{} `json:"metadata,omitempty"` + TwoPCState interface{} `json:"two_pc_state,omitempty"` + }{ + BackupID: backupID, + Type: "incremental", + StartTime: startTime, + WALStart: lastIndex, + WALEnd: currentIndex, + ParentID: lastFullBackup.ID, + Entries: walEntries, + Metadata: map[string]interface{}{ + "version": "1.0", + "created_by": "backup_scheduler", + "entries_count": len(walEntries), + "parent_backup": lastFullBackup.ID, + "parent_time": lastFullBackup.StartTime, + "two_pc_pending_count": twoPCCount, + "include_two_pc": bs.config.IncludeTwoPCState, + }, + TwoPCState: map[string]interface{}{ + "pending_count": twoPCCount, + "timestamp": time.Now().UnixMilli(), + }, + } + + // Если есть 2PC транзакции, добавляем их ID + if twoPCCount > 0 && bs.config.IncludeTwoPCState { + txIDs := make([]string, 0, twoPCCount) + for _, tx := range pending2PC { + txIDs = append(txIDs, fmt.Sprintf("%d", tx.ID)) + } + if twoPCState, ok := backupData.TwoPCState.(map[string]interface{}); ok { + twoPCState["transaction_ids"] = txIDs + } + } + + data, err := json.Marshal(backupData) + if err != nil { + return fmt.Errorf("failed to marshal backup data: %v", err) + } + + // Проверяем размер бэкапа + if bs.config.MaxBackupSizeMB > 0 && int64(len(data)) > bs.config.MaxBackupSizeMB*1024*1024 { + bs.logger.Info("Incremental backup too large, performing full backup instead") + return bs.performFullBackup(scheduleID) + } + + // Сохраняем инкрементальный бэкап + backupPath := fmt.Sprintf("inc/%s/%s.backup", + time.Now().Format("2006-01-02"), backupID) + + if err := bs.storage.SaveBackup(backupPath, data); err != nil { + return fmt.Errorf("failed to save incremental backup: %v", err) + } + + // Обновляем последний индекс + if err := bs.walReader.SetLastBackupIndex(currentIndex); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err)) + } + + hash := sha256.Sum256(data) + checksum := hex.EncodeToString(hash[:]) + + endTime := time.Now().UnixMilli() + + backupInfo := &BackupInfo{ + ID: backupID, + ScheduleID: scheduleID, + Type: IncrementalBackup, + StartTime: startTime, + EndTime: endTime, + Status: "completed", + SizeBytes: int64(len(data)), + Path: backupPath, + WALStart: lastIndex, + WALEnd: currentIndex, + ParentID: lastFullBackup.ID, + Checksum: checksum, + TwoPCCount: twoPCCount, + } + + bs.mu.Lock() + bs.backups[backupID] = backupInfo + bs.mu.Unlock() + + if err := bs.saveBackupHistory(); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err)) + } + + bs.updateScheduleLastRun(scheduleID, startTime) + + duration := (endTime - startTime) / 1000 + bs.logger.Info(fmt.Sprintf("Incremental backup %s completed in %d seconds, %d entries, size=%d bytes, pending_2pc=%d", + backupID, duration, len(walEntries), backupInfo.SizeBytes, twoPCCount)) + + return nil +} + +// findLastFullBackup находит последний успешный полный бэкап +func (bs *BackupScheduler) findLastFullBackup() *BackupInfo { + bs.mu.RLock() + defer bs.mu.RUnlock() + + var lastFull *BackupInfo + for _, backup := range bs.backups { + if backup.Type == FullBackup && backup.Status == "completed" { + if lastFull == nil || backup.EndTime > lastFull.EndTime { + lastFull = backup + } + } + } + return lastFull +} + +// updateScheduleLastRun обновляет время последнего запуска расписания +func (bs *BackupScheduler) updateScheduleLastRun(scheduleID string, timestamp int64) { + bs.mu.Lock() + defer bs.mu.Unlock() + + if schedule, exists := bs.schedules[scheduleID]; exists { + schedule.LastRun = timestamp + schedule.UpdatedAt = time.Now().UnixMilli() + if err := bs.saveSchedules(); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to save schedules after update: %v", err)) + } + } +} + +// cleanupOldBackups удаляет старые бэкапы +func (bs *BackupScheduler) cleanupOldBackups(scheduleID string) { + bs.mu.Lock() + defer bs.mu.Unlock() + + schedule, exists := bs.schedules[scheduleID] + if !exists { + return + } + + cutoffTime := time.Now().AddDate(0, 0, -schedule.RetentionDays).UnixMilli() + + toDelete := make([]string, 0) + for id, backup := range bs.backups { + if backup.ScheduleID == scheduleID && backup.StartTime < cutoffTime { + toDelete = append(toDelete, id) + // Удаляем файл бэкапа + if err := bs.storage.DeleteBackup(backup.Path); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to delete backup file %s: %v", backup.Path, err)) + } + } + } + + for _, id := range toDelete { + delete(bs.backups, id) + } + + if len(toDelete) > 0 { + if err := bs.saveBackupHistory(); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to save backup history after cleanup: %v", err)) + } + bs.logger.Info(fmt.Sprintf("Cleaned up %d old backups for schedule %s", len(toDelete), scheduleID)) + } +} + +// ListBackups возвращает список бэкапов +func (bs *BackupScheduler) ListBackups() []*BackupInfo { + bs.mu.RLock() + defer bs.mu.RUnlock() + + backups := make([]*BackupInfo, 0, len(bs.backups)) + for _, b := range bs.backups { + backups = append(backups, b) + } + + // Сортируем по времени (новейшие первыми) + sort.Slice(backups, func(i, j int) bool { + return backups[i].StartTime > backups[j].StartTime + }) + + return backups +} + +// ListSchedules возвращает список расписаний +func (bs *BackupScheduler) ListSchedules() []*BackupSchedule { + bs.mu.RLock() + defer bs.mu.RUnlock() + + schedules := make([]*BackupSchedule, 0, len(bs.schedules)) + for _, s := range bs.schedules { + schedules = append(schedules, s) + } + return schedules +} + +// GetBackup возвращает информацию о бэкапе +func (bs *BackupScheduler) GetBackup(backupID string) *BackupInfo { + bs.mu.RLock() + defer bs.mu.RUnlock() + return bs.backups[backupID] +} + +// Restore восстанавливает данные из бэкапа +func (bs *BackupScheduler) Restore(backupID string) error { + if !bs.running.CompareAndSwap(false, true) { + return fmt.Errorf("restore already in progress") + } + defer bs.running.Store(false) + + backup := bs.GetBackup(backupID) + if backup == nil { + return fmt.Errorf("backup not found: %s", backupID) + } + + bs.logger.Info(fmt.Sprintf("Starting restore from backup %s", backupID)) + + // Проверяем существование файла бэкапа + if !bs.storage.BackupExists(backup.Path) { + return fmt.Errorf("backup file not found: %s", backup.Path) + } + + data, err := bs.storage.LoadBackup(backup.Path) + if err != nil { + return fmt.Errorf("failed to load backup: %v", err) + } + + // Проверяем контрольную сумму + hash := sha256.Sum256(data) + checksum := hex.EncodeToString(hash[:]) + if checksum != backup.Checksum { + return fmt.Errorf("backup checksum mismatch, data may be corrupted") + } + + if backup.Type == FullBackup { + // Восстановление из полного бэкапа + bs.logger.Info("Restoring from full backup...") + + var fullData struct { + BackupID string `json:"backup_id"` + Type string `json:"type"` + Entries []WALEntry `json:"entries"` + Metadata interface{} `json:"metadata"` + TwoPCState interface{} `json:"two_pc_state,omitempty"` + } + + if err := json.Unmarshal(data, &fullData); err != nil { + return fmt.Errorf("failed to parse full backup data: %v", err) + } + + bs.logger.Info(fmt.Sprintf("Full backup contains %d WAL entries", len(fullData.Entries))) + + // ========== ДОБАВЛЕНО: Восстановление 2PC состояния ========== + if fullData.TwoPCState != nil && bs.config.IncludeTwoPCState { + bs.logger.Info("Restoring 2PC state from backup...") + // Здесь логика восстановления 2PC состояния + if twoPCState, ok := fullData.TwoPCState.(map[string]interface{}); ok { + if pendingCount, ok := twoPCState["pending_count"].(float64); ok && pendingCount > 0 { + bs.logger.Info(fmt.Sprintf("Found %d pending 2PC transactions in backup", int(pendingCount))) + // Восстанавливаем pending транзакции в менеджере + if bs.txManager != nil { + // Здесь можно восстановить состояние 2PC + // В реальной реализации нужно восстановить транзакции из WAL + } + } + } + } + + // Здесь реальное восстановление данных + + } else { + // Восстановление из инкрементального бэкапа + bs.logger.Info("Restoring from incremental backup...") + + // Сначала восстанавливаем полный бэкап-родитель + if backup.ParentID == "" { + return fmt.Errorf("incremental backup has no parent ID") + } + + // Проверяем существование родительского бэкапа + parentBackup := bs.GetBackup(backup.ParentID) + if parentBackup == nil { + return fmt.Errorf("parent backup %s not found", backup.ParentID) + } + + if err := bs.Restore(backup.ParentID); err != nil { + return fmt.Errorf("failed to restore parent backup: %v", err) + } + + var incData struct { + BackupID string `json:"backup_id"` + Type string `json:"type"` + WALStart uint64 `json:"wal_start"` + WALEnd uint64 `json:"wal_end"` + Entries []WALEntry `json:"entries"` + TwoPCState interface{} `json:"two_pc_state,omitempty"` + } + + if err := json.Unmarshal(data, &incData); err != nil { + return fmt.Errorf("failed to parse incremental backup data: %v", err) + } + + bs.logger.Info(fmt.Sprintf("Applying %d incremental changes (WAL %d -> %d)", + len(incData.Entries), incData.WALStart, incData.WALEnd)) + + // ========== ДОБАВЛЕНО: Восстановление 2PC состояния из инкремента ========== + if incData.TwoPCState != nil && bs.config.IncludeTwoPCState { + bs.logger.Info("Restoring 2PC state from incremental backup...") + if twoPCState, ok := incData.TwoPCState.(map[string]interface{}); ok { + if pendingCount, ok := twoPCState["pending_count"].(float64); ok && pendingCount > 0 { + bs.logger.Info(fmt.Sprintf("Found %d pending 2PC transactions in incremental backup", int(pendingCount))) + } + } + } + + // Здесь применяем инкрементальные изменения + } + + bs.logger.Info(fmt.Sprintf("Restore from backup %s completed", backupID)) + return nil +} + +// Start запускает планировщик +func (bs *BackupScheduler) Start() { + bs.wg.Add(1) + go bs.schedulerLoop() + if bs.logger != nil { + bs.logger.Info("Backup scheduler started") + } +} + +// schedulerLoop основной цикл планировщика +func (bs *BackupScheduler) schedulerLoop() { + defer bs.wg.Done() + + ticker := time.NewTicker(60 * time.Second) + defer ticker.Stop() + + for { + select { + case <-bs.stopChan: + return + case <-ticker.C: + bs.checkSchedules() + } + } +} + +// checkSchedules проверяет расписания и запускает бэкапы +func (bs *BackupScheduler) checkSchedules() { + bs.mu.RLock() + schedules := make([]*BackupSchedule, 0, len(bs.schedules)) + for _, s := range bs.schedules { + if s.Enabled { + schedules = append(schedules, s) + } + } + bs.mu.RUnlock() + + now := time.Now() + for _, schedule := range schedules { + // Простая проверка расписания (в реальности используется cron-парсер) + if schedule.NextRun == 0 || schedule.NextRun <= now.UnixMilli() { + // Вычисляем следующее время (упрощённо - через 24 часа) + schedule.NextRun = now.Add(24 * time.Hour).UnixMilli() + + bs.mu.Lock() + bs.schedules[schedule.ID] = schedule + bs.mu.Unlock() + + if err := bs.saveSchedules(); err != nil { + bs.logger.Warn(fmt.Sprintf("Failed to save schedule after update: %v", err)) + } + + // Запускаем бэкап в отдельной горутине + scheduleID := schedule.ID + backupType := schedule.Type + go func() { + if backupType == FullBackup { + bs.performFullBackup(scheduleID) + } else { + bs.performIncrementalBackup(scheduleID) + } + }() + } + } +} + +// Stop останавливает планировщик +func (bs *BackupScheduler) Stop() { + close(bs.stopChan) + bs.wg.Wait() + if bs.logger != nil { + bs.logger.Info("Backup scheduler stopped") + } +} + +// GetStats возвращает статистику планировщика +func (bs *BackupScheduler) GetStats() map[string]interface{} { + bs.mu.RLock() + defer bs.mu.RUnlock() + + fullCount := 0 + incCount := 0 + totalSize := int64(0) + total2PC := 0 + + for _, b := range bs.backups { + if b.Type == FullBackup { + fullCount++ + } else { + incCount++ + } + totalSize += b.SizeBytes + total2PC += b.TwoPCCount + } + + return map[string]interface{}{ + "total_backups": len(bs.backups), + "full_backups": fullCount, + "incremental_backups": incCount, + "total_size_bytes": totalSize, + "schedules": len(bs.schedules), + "is_running": bs.running.Load(), + "backup_dir": bs.config.BackupDir, + "compress_enabled": bs.config.CompressEnabled, + "incremental_enabled": bs.config.EnableIncremental, + "include_two_pc_state": bs.config.IncludeTwoPCState, + "total_2pc_transactions": total2PC, + } +}