From 75506b9fb7f26c90e852bcad94d6e14ec1f7bba8 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: Wed, 8 Jul 2026 20:44:07 +0000 Subject: [PATCH] Delete internal/storage/persistence.go --- internal/storage/persistence.go | 980 -------------------------------- 1 file changed, 980 deletions(-) delete mode 100644 internal/storage/persistence.go diff --git a/internal/storage/persistence.go b/internal/storage/persistence.go deleted file mode 100644 index 6d0a1d6..0000000 --- a/internal/storage/persistence.go +++ /dev/null @@ -1,980 +0,0 @@ -/* - * 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/storage/persistence.go -// Назначение: Персистентное хранение данных на диске с поддержкой checkpoint и recovery - -package storage - -import ( - "bytes" - "compress/gzip" - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "os" - "path/filepath" - "sort" - "strings" - "sync" - "sync/atomic" - "time" -) - -// ========== WALReader интерфейс и реализация ========== - -// 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) -} - -// walReaderImpl реализация WALReader для работы с файлами WAL -type walReaderImpl struct { - segmentsDir string - lastBackupIndex atomic.Uint64 - mu sync.RWMutex - logger LoggerInterface -} - -// NewWALReaderImpl создаёт новый WALReader -func NewWALReaderImpl(segmentsDir string, logger LoggerInterface) WALReader { - wr := &walReaderImpl{ - segmentsDir: segmentsDir, - logger: logger, - } - wr.loadLastBackupIndex() - return wr -} - -// 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) - } -} - -// 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) - - pos := 0 - for pos < len(data) { - if pos+4 > len(data) { - break - } - - 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 -} - -// ========== PersistenceConfig ========== - -// PersistenceConfig конфигурация персистентного хранения -type PersistenceConfig struct { - DataDir string `json:"data_dir"` - CheckpointInterval time.Duration `json:"checkpoint_interval"` - MaxCheckpoints int `json:"max_checkpoints"` - CompressEnabled bool `json:"compress_enabled"` - SyncWrites bool `json:"sync_writes"` - WalPath string `json:"wal_path"` - UseWalForCheckpoint bool `json:"use_wal_for_checkpoint"` - AtomicWrites bool `json:"atomic_writes"` -} - -// DefaultPersistenceConfig возвращает конфигурацию по умолчанию -func DefaultPersistenceConfig() *PersistenceConfig { - return &PersistenceConfig{ - DataDir: "futriis_data", - CheckpointInterval: 5 * time.Minute, - MaxCheckpoints: 10, - CompressEnabled: true, - SyncWrites: true, - WalPath: "futriis.wal", - UseWalForCheckpoint: true, - AtomicWrites: true, - } -} - -// DatabaseSnapshot представляет снапшот базы данных -type DatabaseSnapshot struct { - Name string `json:"name"` - Collections map[string]interface{} `json:"collections"` - CreatedAt int64 `json:"created_at"` - Version uint64 `json:"version"` - Checksum string `json:"checksum"` - WalLSN uint64 `json:"wal_lsn,omitempty"` - WalSegment string `json:"wal_segment,omitempty"` -} - -// CheckpointRecoveryInfo содержит информацию о восстановлении из чекпоинта -type CheckpointRecoveryInfo struct { - CheckpointTime int64 `json:"checkpoint_time"` - CheckpointVersion uint64 `json:"checkpoint_version"` - WalLSN uint64 `json:"wal_lsn"` - AppliedWalCount int `json:"applied_wal_count"` - RestoredDocs int64 `json:"restored_docs"` - RestoredColls int `json:"restored_collections"` - DurationMs int64 `json:"duration_ms"` - Success bool `json:"success"` - Error string `json:"error,omitempty"` -} - -// PersistenceManager управляет персистентным хранением -type PersistenceManager struct { - config *PersistenceConfig - storage *Storage - logger LoggerInterface - mu sync.RWMutex - stopChan chan struct{} - wg sync.WaitGroup - lastCheckpoint int64 - checkpointID atomic.Uint64 - walManager interface{} - walReader WALReader - recoveryInfo *CheckpointRecoveryInfo - isRestoring atomic.Bool -} - -// NewPersistenceManager создаёт новый менеджер персистентности -func NewPersistenceManager(config *PersistenceConfig, storage *Storage, logger LoggerInterface) *PersistenceManager { - if config == nil { - config = DefaultPersistenceConfig() - } - - pm := &PersistenceManager{ - config: config, - storage: storage, - logger: logger, - stopChan: make(chan struct{}), - lastCheckpoint: time.Now().UnixMilli(), - } - - if err := os.MkdirAll(config.DataDir, 0755); err != nil { - if logger != nil { - logger.Error(fmt.Sprintf("Failed to create data directory: %v", err)) - } - } - - if config.UseWalForCheckpoint && config.WalPath != "" { - pm.walReader = NewWALReaderImpl(filepath.Dir(config.WalPath), logger) - } - - return pm -} - -// SetWALManager устанавливает WAL менеджер для синхронизации чекпоинтов -func (pm *PersistenceManager) SetWALManager(walManager interface{}) { - pm.mu.Lock() - defer pm.mu.Unlock() - pm.walManager = walManager - - if walManager != nil && pm.config.UseWalForCheckpoint { - pm.walReader = NewWALReaderImpl(filepath.Dir(pm.config.WalPath), pm.logger) - } -} - -// Start запускает фоновое сохранение чекпоинтов -func (pm *PersistenceManager) Start() { - pm.wg.Add(1) - go pm.checkpointLoop() - - if pm.logger != nil { - pm.logger.Info(fmt.Sprintf("Persistence manager started, data dir: %s", pm.config.DataDir)) - } -} - -// Stop останавливает менеджер -func (pm *PersistenceManager) Stop() { - close(pm.stopChan) - pm.wg.Wait() - - pm.SaveAll() - - if pm.logger != nil { - pm.logger.Info("Persistence manager stopped") - } -} - -// checkpointLoop периодически создаёт чекпоинты -func (pm *PersistenceManager) checkpointLoop() { - defer pm.wg.Done() - - ticker := time.NewTicker(pm.config.CheckpointInterval) - defer ticker.Stop() - - for { - select { - case <-pm.stopChan: - return - case <-ticker.C: - if pm.isRestoring.Load() { - if pm.logger != nil { - pm.logger.Info("Skipping checkpoint during restore") - } - continue - } - if err := pm.SaveAll(); err != nil { - if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to save checkpoint: %v", err)) - } - } else { - pm.cleanupOldCheckpoints() - } - } - } -} - -// SaveDatabase сохраняет базу данных на диск с атомарной записью -func (pm *PersistenceManager) SaveDatabase(dbName string) error { - if pm.isRestoring.Load() { - return fmt.Errorf("cannot save checkpoint during restore") - } - - pm.mu.Lock() - defer pm.mu.Unlock() - - db, err := pm.storage.GetDatabase(dbName) - if err != nil { - return fmt.Errorf("database not found: %s", dbName) - } - - var walLSN uint64 - var walSegment string - - if pm.config.UseWalForCheckpoint && pm.walManager != nil { - if segWal, ok := pm.walManager.(*SegmentedWALManager); ok { - if segWal.currentSegment != nil { - segWal.currentSegment.Writer.Flush() - RealFsync(segWal.currentSegment.File) - walLSN = segWal.currentSegment.EndLSN - walSegment = fmt.Sprintf("%d", segWal.currentSegment.ID) - } - } - } - - snapshot := &DatabaseSnapshot{ - Name: dbName, - Collections: make(map[string]interface{}), - CreatedAt: time.Now().UnixMilli(), - Version: pm.checkpointID.Add(1), - WalLSN: walLSN, - WalSegment: walSegment, - } - - for _, collName := range db.ListCollections() { - coll, err := db.GetCollection(collName) - if err != nil { - continue - } - - docs := coll.GetAllDocuments() - collData := make([]map[string]interface{}, 0, len(docs)) - for _, doc := range docs { - collData = append(collData, map[string]interface{}{ - "_id": doc.ID, - "fields": doc.GetFields(), - "created_at": doc.CreatedAt, - "updated_at": doc.UpdatedAt, - "deleted_at": doc.DeletedAt, - "version": doc.Version, - }) - } - snapshot.Collections[collName] = collData - } - - data, err := json.Marshal(snapshot) - if err != nil { - return err - } - - // Вычисляем контрольную сумму - hash := sha256.Sum256(data) - snapshot.Checksum = hex.EncodeToString(hash[:]) - - // Пере-сериализуем с контрольной суммой - data, err = json.Marshal(snapshot) - if err != nil { - return err - } - - if pm.config.CompressEnabled { - data, err = pm.compress(data) - if err != nil { - return err - } - } - - // Атомарная запись с использованием временного файла - filename := pm.getSnapshotFilename(dbName, snapshot.Version) - if err := pm.atomicWriteFile(filename, data); err != nil { - return err - } - - pm.lastCheckpoint = snapshot.CreatedAt - - if pm.logger != nil { - pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes, checksum: %s, WAL LSN: %d)", - dbName, snapshot.Version, len(data), snapshot.Checksum[:8], walLSN)) - } - - return nil -} - -// atomicWriteFile выполняет атомарную запись файла -func (pm *PersistenceManager) atomicWriteFile(filename string, data []byte) error { - if pm.config.AtomicWrites { - // Создаём временный файл в той же директории - tempFile := filename + ".tmp" - if err := os.WriteFile(tempFile, data, 0644); err != nil { - return fmt.Errorf("failed to write temp file: %v", err) - } - - // Синхронизируем временный файл - if pm.config.SyncWrites { - if f, err := os.OpenFile(tempFile, os.O_RDWR, 0644); err == nil { - RealFsync(f) - f.Close() - } - } - - // Атомарное переименование - if err := os.Rename(tempFile, filename); err != nil { - os.Remove(tempFile) - return fmt.Errorf("failed to rename temp file: %v", err) - } - - // Синхронизируем директорию - FsyncDir(pm.config.DataDir) - - return nil - } - - // Обычная запись (не атомарная) - return os.WriteFile(filename, data, 0644) -} - -// SaveAll сохраняет все базы данных -func (pm *PersistenceManager) SaveAll() error { - if pm.isRestoring.Load() { - return fmt.Errorf("cannot save checkpoints during restore") - } - - databases := pm.storage.ListDatabases() - - var lastErr error - for _, dbName := range databases { - if err := pm.SaveDatabase(dbName); err != nil { - lastErr = err - if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to save database %s: %v", dbName, err)) - } - } - } - - return lastErr -} - -// LoadDatabase загружает базу данных с диска с применением WAL -func (pm *PersistenceManager) LoadDatabase(dbName string) error { - pm.isRestoring.Store(true) - defer pm.isRestoring.Store(false) - - pm.mu.Lock() - defer pm.mu.Unlock() - - startTime := time.Now() - - recoveryInfo := &CheckpointRecoveryInfo{ - CheckpointTime: time.Now().UnixMilli(), - Success: false, - } - - snapshot, err := pm.findLatestSnapshot(dbName) - if err != nil { - recoveryInfo.Error = err.Error() - pm.recoveryInfo = recoveryInfo - return err - } - - if snapshot == nil { - recoveryInfo.Error = "no snapshot found" - pm.recoveryInfo = recoveryInfo - return fmt.Errorf("no snapshot found for database: %s", dbName) - } - - recoveryInfo.CheckpointVersion = snapshot.Version - recoveryInfo.WalLSN = snapshot.WalLSN - - // Проверяем целостность чекпоинта - if err := pm.verifySnapshot(snapshot); err != nil { - recoveryInfo.Error = err.Error() - pm.recoveryInfo = recoveryInfo - return fmt.Errorf("snapshot integrity check failed: %v", err) - } - - // Создаём базу данных если не существует - if !pm.storage.ExistsDatabase(dbName) { - if err := pm.storage.CreateDatabase(dbName); err != nil { - recoveryInfo.Error = err.Error() - pm.recoveryInfo = recoveryInfo - return err - } - } - - db, err := pm.storage.GetDatabase(dbName) - if err != nil { - recoveryInfo.Error = err.Error() - pm.recoveryInfo = recoveryInfo - return err - } - - // Восстанавливаем коллекции - restoredColls := 0 - restoredDocs := int64(0) - - for collName, collDataRaw := range snapshot.Collections { - collData, ok := collDataRaw.([]interface{}) - if !ok { - continue - } - - // Удаляем существующую коллекцию если есть - if _, err := db.GetCollection(collName); err == nil { - db.DropCollection(collName) - } - - if err := db.CreateCollection(collName); err != nil { - if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to create collection %s: %v", collName, err)) - } - continue - } - - coll, err := db.GetCollection(collName) - if err != nil { - continue - } - - restoredColls++ - - for _, docRaw := range collData { - docMap, ok := docRaw.(map[string]interface{}) - if !ok { - continue - } - - docID, ok := docMap["_id"].(string) - if !ok { - continue - } - - doc := NewDocumentWithID(docID) - - if fields, ok := docMap["fields"].(map[string]interface{}); ok { - for k, v := range fields { - doc.SetField(k, v) - } - } - - if createdAt, ok := docMap["created_at"].(float64); ok { - doc.CreatedAt = int64(createdAt) - } else if createdAt, ok := docMap["created_at"].(int64); ok { - doc.CreatedAt = createdAt - } - - if updatedAt, ok := docMap["updated_at"].(float64); ok { - doc.UpdatedAt = int64(updatedAt) - } else if updatedAt, ok := docMap["updated_at"].(int64); ok { - doc.UpdatedAt = updatedAt - } - - if deletedAt, ok := docMap["deleted_at"].(float64); ok { - doc.DeletedAt = int64(deletedAt) - } else if deletedAt, ok := docMap["deleted_at"].(int64); ok { - doc.DeletedAt = deletedAt - } - - if version, ok := docMap["version"].(float64); ok { - doc.Version = uint64(version) - } else if version, ok := docMap["version"].(uint64); ok { - doc.Version = version - } - - if err := coll.Insert(doc); err != nil { - if pm.logger != nil { - pm.logger.Warn(fmt.Sprintf("Failed to restore document %s: %v", docID, err)) - } - continue - } - restoredDocs++ - } - } - - recoveryInfo.RestoredDocs = restoredDocs - recoveryInfo.RestoredColls = restoredColls - - // Применяем WAL записи после чекпоинта - appliedWalCount := 0 - if pm.config.UseWalForCheckpoint && snapshot.WalLSN > 0 && pm.walReader != nil { - currentLSN, err := pm.walReader.GetCurrentIndex() - if err == nil && currentLSN > snapshot.WalLSN { - if pm.logger != nil { - pm.logger.Info(fmt.Sprintf("Database %s: Applying %d WAL entries after checkpoint (LSN %d -> %d)", - dbName, currentLSN-snapshot.WalLSN, snapshot.WalLSN, currentLSN)) - } - - entries, err := pm.walReader.ReadSince(snapshot.WalLSN) - if err != nil { - if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to read WAL entries: %v", err)) - } - } else { - for _, entry := range entries { - if err := pm.applyWALEntry(db, entry); err != nil { - if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to apply WAL entry %d: %v", entry.Index, err)) - } - continue - } - appliedWalCount++ - } - } - } - } - - recoveryInfo.AppliedWalCount = appliedWalCount - recoveryInfo.DurationMs = time.Since(startTime).Milliseconds() - recoveryInfo.Success = true - - pm.recoveryInfo = recoveryInfo - - if pm.logger != nil { - pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d, WAL LSN: %d, applied %d WAL entries, %d docs, %d colls, duration: %dms)", - dbName, snapshot.Version, snapshot.WalLSN, appliedWalCount, restoredDocs, restoredColls, recoveryInfo.DurationMs)) - } - - return nil -} - -// applyWALEntry применяет одну запись WAL к базе данных -func (pm *PersistenceManager) applyWALEntry(db *Database, entry WALEntry) error { - var txRecord struct { - ID uint64 `json:"id"` - State int32 `json:"state"` - Timestamp int64 `json:"timestamp"` - Operations []struct { - Type string `json:"type"` - Database string `json:"database"` - Collection string `json:"collection"` - DocumentID string `json:"document_id"` - Data map[string]interface{} `json:"data"` - Version uint64 `json:"version"` - } `json:"operations"` - } - - if err := json.Unmarshal(entry.Data, &txRecord); err != nil { - return err - } - - // Применяем только закоммиченные транзакции - if txRecord.State != 1 { // TransactionCommitted - return nil - } - - for _, op := range txRecord.Operations { - // Пропускаем операции для других баз данных - if op.Database != db.name { - continue - } - - coll, err := db.GetCollection(op.Collection) - if err != nil { - continue - } - - switch op.Type { - case "insert": - doc := NewDocumentWithID(op.DocumentID) - for k, v := range op.Data { - doc.SetField(k, v) - } - doc.Version = op.Version - if err := coll.Insert(doc); err != nil { - return err - } - - case "update": - if err := coll.Update(op.DocumentID, op.Data); err != nil { - return err - } - - case "delete": - if err := coll.Delete(op.DocumentID); err != nil { - return err - } - } - } - - return nil -} - -// verifySnapshot проверяет целостность снапшота -func (pm *PersistenceManager) verifySnapshot(snapshot *DatabaseSnapshot) error { - if snapshot.Checksum == "" { - return fmt.Errorf("snapshot has no checksum") - } - - // Пересоздаём данные без контрольной суммы для проверки - tempSnapshot := &DatabaseSnapshot{ - Name: snapshot.Name, - Collections: snapshot.Collections, - CreatedAt: snapshot.CreatedAt, - Version: snapshot.Version, - WalLSN: snapshot.WalLSN, - WalSegment: snapshot.WalSegment, - } - - data, err := json.Marshal(tempSnapshot) - if err != nil { - return fmt.Errorf("failed to marshal for checksum: %v", err) - } - - hash := sha256.Sum256(data) - checksum := hex.EncodeToString(hash[:]) - - if checksum != snapshot.Checksum { - return fmt.Errorf("checksum mismatch: expected %s, got %s", snapshot.Checksum[:8], checksum[:8]) - } - - return nil -} - -// LoadAll загружает все базы данных с диска -func (pm *PersistenceManager) LoadAll() error { - files, err := filepath.Glob(filepath.Join(pm.config.DataDir, "snapshot_*.json*")) - if err != nil { - return err - } - - databases := make(map[string]bool) - for _, file := range files { - base := filepath.Base(file) - parts := strings.Split(base, "_") - if len(parts) >= 2 { - dbName := parts[1] - databases[dbName] = true - } - } - - for dbName := range databases { - if err := pm.LoadDatabase(dbName); err != nil { - if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to load database %s: %v", dbName, err)) - } - } - } - - return nil -} - -// getSnapshotFilename возвращает имя файла для снапшота -func (pm *PersistenceManager) getSnapshotFilename(dbName string, version uint64) string { - filename := fmt.Sprintf("snapshot_%s_%d.json", dbName, version) - if pm.config.CompressEnabled { - filename += ".gz" - } - return filepath.Join(pm.config.DataDir, filename) -} - -// findLatestSnapshot находит последний снапшот базы данных -func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapshot, error) { - pattern := filepath.Join(pm.config.DataDir, fmt.Sprintf("snapshot_%s_*.json*", dbName)) - files, err := filepath.Glob(pattern) - if err != nil { - return nil, err - } - - if len(files) == 0 { - return nil, nil - } - - sort.Slice(files, func(i, j int) bool { - infoI, _ := os.Stat(files[i]) - infoJ, _ := os.Stat(files[j]) - if infoI == nil || infoJ == nil { - return false - } - return infoI.ModTime().After(infoJ.ModTime()) - }) - - data, err := os.ReadFile(files[0]) - if err != nil { - return nil, err - } - - if pm.config.CompressEnabled && strings.HasSuffix(files[0], ".gz") { - data, err = pm.decompress(data) - if err != nil { - return nil, err - } - } - - var snapshot DatabaseSnapshot - if err := json.Unmarshal(data, &snapshot); err != nil { - return nil, err - } - - return &snapshot, nil -} - -// compress сжимает данные с помощью gzip -func (pm *PersistenceManager) compress(data []byte) ([]byte, error) { - var buf bytes.Buffer - w := gzip.NewWriter(&buf) - - if _, err := w.Write(data); err != nil { - return nil, fmt.Errorf("failed to write compressed data: %v", err) - } - - if err := w.Close(); err != nil { - return nil, fmt.Errorf("failed to close gzip writer: %v", err) - } - - return buf.Bytes(), nil -} - -// decompress разжимает данные с помощью gzip -func (pm *PersistenceManager) decompress(data []byte) ([]byte, error) { - reader, err := gzip.NewReader(bytes.NewReader(data)) - if err != nil { - return nil, fmt.Errorf("failed to create gzip reader: %v", err) - } - defer reader.Close() - - var buf bytes.Buffer - if _, err := buf.ReadFrom(reader); err != nil { - return nil, fmt.Errorf("failed to decompress data: %v", err) - } - - return buf.Bytes(), nil -} - -// cleanupOldCheckpoints удаляет старые чекпоинты -func (pm *PersistenceManager) cleanupOldCheckpoints() { - pattern := filepath.Join(pm.config.DataDir, "snapshot_*.json*") - files, err := filepath.Glob(pattern) - if err != nil { - return - } - - if len(files) <= pm.config.MaxCheckpoints { - return - } - - sort.Slice(files, func(i, j int) bool { - infoI, _ := os.Stat(files[i]) - infoJ, _ := os.Stat(files[j]) - if infoI == nil || infoJ == nil { - return false - } - return infoI.ModTime().Before(infoJ.ModTime()) - }) - - toDelete := files[:len(files)-pm.config.MaxCheckpoints] - for _, f := range toDelete { - os.Remove(f) - if pm.logger != nil { - pm.logger.Debug(fmt.Sprintf("Removed old checkpoint: %s", f)) - } - } -} - -// GetLastCheckpointInfo возвращает информацию о последнем чекпоинте -func (pm *PersistenceManager) GetLastCheckpointInfo() map[string]interface{} { - pm.mu.RLock() - defer pm.mu.RUnlock() - - info := map[string]interface{}{ - "last_checkpoint_time": pm.lastCheckpoint, - "last_checkpoint_time_str": time.UnixMilli(pm.lastCheckpoint).Format("2006-01-02 15:04:05.000"), - "checkpoint_id": pm.checkpointID.Load(), - "wal_path": pm.config.WalPath, - "use_wal": pm.config.UseWalForCheckpoint, - "atomic_writes": pm.config.AtomicWrites, - "is_restoring": pm.isRestoring.Load(), - } - - if pm.recoveryInfo != nil { - info["last_recovery"] = pm.recoveryInfo - } - - return info -} - -// GetRecoveryInfo возвращает информацию о последнем восстановлении -func (pm *PersistenceManager) GetRecoveryInfo() *CheckpointRecoveryInfo { - pm.mu.RLock() - defer pm.mu.RUnlock() - return pm.recoveryInfo -} - -// IsRestoring возвращает статус восстановления -func (pm *PersistenceManager) IsRestoring() bool { - return pm.isRestoring.Load() -} \ No newline at end of file