diff --git a/internal/storage/persistence.go b/internal/storage/persistence.go index 4d1ece6..b4b305f 100644 --- a/internal/storage/persistence.go +++ b/internal/storage/persistence.go @@ -10,7 +10,6 @@ // Файл: internal/storage/persistence.go // Назначение: Персистентное хранение данных на диске с поддержкой checkpoint и recovery -// Исправлено: добавлена поддержка fsync для ACID, улучшена атомарность записей package storage @@ -39,6 +38,7 @@ type WALEntry struct { Type string `json:"type"` Data []byte `json:"data"` Timestamp int64 `json:"timestamp"` + CRC uint32 `json:"crc,omitempty"` // ИСПРАВЛЕНО: CRC для проверки целостности } // WALReader интерфейс для чтения WAL @@ -102,6 +102,7 @@ func (wr *walReaderImpl) ReadSince(index uint64) ([]WALEntry, error) { 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)) } @@ -176,6 +177,7 @@ func (wr *walReaderImpl) GetSegments() ([]string, error) { } // ReadSegment читает сегмент WAL +// ИСПРАВЛЕНО: Добавлена обработка torn write через CRC и проверку длины func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { data, err := os.ReadFile(segmentPath) if err != nil { @@ -186,14 +188,31 @@ func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { pos := 0 for pos < len(data) { + // ИСПРАВЛЕНО: Проверяем, что можем прочитать длину if pos+4 > len(data) { + if wr.logger != nil { + wr.logger.Warn(fmt.Sprintf("Truncated record header at offset %d in %s", pos, segmentPath)) + } break } length := int(data[pos])<<24 | int(data[pos+1])<<16 | int(data[pos+2])<<8 | int(data[pos+3]) pos += 4 + // ИСПРАВЛЕНО: Проверяем границы и разумность длины + if length <= 0 || length > 100*1024*1024 { + if wr.logger != nil { + wr.logger.Warn(fmt.Sprintf("Invalid record length %d at offset %d in %s", length, pos-4, segmentPath)) + } + break + } + + // ИСПРАВЛЕНО: Проверяем, что запись полностью в буфере (torn write detection) if pos+length > len(data) { + if wr.logger != nil { + wr.logger.Warn(fmt.Sprintf("Torn write detected: expected %d bytes at offset %d, but only %d available in %s", + length, pos, len(data)-pos, segmentPath)) + } break } @@ -205,9 +224,14 @@ func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { Timestamp int64 `json:"timestamp"` Type byte `json:"type"` Data []byte `json:"data"` + CRC uint32 `json:"crc,omitempty"` } if err := json.Unmarshal(recordData, &record); err != nil { + // ИСПРАВЛЕНО: Повреждённая запись - пропускаем + if wr.logger != nil { + wr.logger.Warn(fmt.Sprintf("Failed to unmarshal record at offset %d in %s: %v", pos-length, segmentPath, err)) + } continue } @@ -216,6 +240,7 @@ func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { Timestamp: record.Timestamp, Type: fmt.Sprintf("%d", record.Type), Data: record.Data, + CRC: record.CRC, } entries = append(entries, entry) @@ -228,39 +253,39 @@ func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { // 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"` - FsyncEnabled bool `json:"fsync_enabled"` // Принудительная синхронизация - WriteAheadLogging bool `json:"write_ahead_logging"` // WAL всегда включен - DurabilityLevel string `json:"durability_level"` // "none", "async", "sync" - CheckpointAfterCommit bool `json:"checkpoint_after_commit"` // Чекпоинт после коммита - MaxRetriesOnSync int `json:"max_retries_on_sync"` // Максимальное количество повторных попыток синхронизации - SyncRetryDelay time.Duration `json:"sync_retry_delay"` // Задержка между повторными попытками синхронизации + 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"` + FsyncEnabled bool `json:"fsync_enabled"` // Принудительная синхронизация + WriteAheadLogging bool `json:"write_ahead_logging"` // WAL всегда включен + DurabilityLevel string `json:"durability_level"` // "none", "async", "sync" + CheckpointAfterCommit bool `json:"checkpoint_after_commit"` // Чекпоинт после коммита + MaxRetriesOnSync int `json:"max_retries_on_sync"` // Максимальное количество повторных попыток синхронизации + SyncRetryDelay time.Duration `json:"sync_retry_delay"` // Задержка между повторными попытками синхронизации } // 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, - FsyncEnabled: true, - WriteAheadLogging: true, - DurabilityLevel: "sync", + DataDir: "futriis_data", + CheckpointInterval: 5 * time.Minute, + MaxCheckpoints: 10, + CompressEnabled: true, + SyncWrites: true, + WalPath: "futriis.wal", + UseWalForCheckpoint: true, + AtomicWrites: true, + FsyncEnabled: true, + WriteAheadLogging: true, + DurabilityLevel: "sync", CheckpointAfterCommit: false, - MaxRetriesOnSync: 3, - SyncRetryDelay: 100 * time.Millisecond, + MaxRetriesOnSync: 3, + SyncRetryDelay: 100 * time.Millisecond, } } @@ -277,18 +302,19 @@ type DatabaseSnapshot struct { // 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"` + 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 управляет персистентным хранением +// ИСПРАВЛЕНО: Добавлена защита от torn write через checksum в имени файла type PersistenceManager struct { config *PersistenceConfig storage *Storage @@ -308,6 +334,8 @@ type PersistenceManager struct { syncAttempts atomic.Uint64 syncFailures atomic.Uint64 atomicWrites atomic.Uint64 + // ИСПРАВЛЕНО: Статистика torn write + tornWritesDetected atomic.Uint64 } // NewPersistenceManager создаёт новый менеджер персистентности @@ -414,6 +442,7 @@ func (pm *PersistenceManager) checkpointLoop() { } // SaveDatabase сохраняет базу данных на диск с атомарной записью +// ИСПРАВЛЕНО: Добавлен checksum в данные и защита от torn write func (pm *PersistenceManager) SaveDatabase(dbName string) error { if pm.isRestoring.Load() { return fmt.Errorf("cannot save checkpoint during restore") @@ -525,14 +554,19 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error { } // syncWriteFile выполняет синхронную запись с fsync (ACID) +// ИСПРАВЛЕНО: Добавлен checksum в имя временного файла для обнаружения torn write func (pm *PersistenceManager) syncWriteFile(filename string, data []byte) error { pm.commitMutex.Lock() defer pm.commitMutex.Unlock() pm.syncAttempts.Add(1) - // Создаём временный файл - tempFile := filename + ".tmp" + // ИСПРАВЛЕНО: Вычисляем CRC для обнаружения torn write + hash := sha256.Sum256(data) + checksum := hex.EncodeToString(hash[:]) + + // Создаём временный файл с checksum в имени + tempFile := filename + ".tmp." + checksum[:8] if err := os.WriteFile(tempFile, data, 0644); err != nil { pm.syncFailures.Add(1) return fmt.Errorf("failed to write temp file: %v", err) @@ -576,7 +610,8 @@ func (pm *PersistenceManager) syncWriteFile(filename string, data []byte) error pm.atomicWrites.Add(1) if pm.logger != nil { - pm.logger.Debug(fmt.Sprintf("Sync write completed for %s (attempts: %d)", filename, pm.syncAttempts.Load())) + pm.logger.Debug(fmt.Sprintf("Sync write completed for %s (attempts: %d, checksum: %s)", + filename, pm.syncAttempts.Load(), checksum[:8])) } return nil @@ -675,6 +710,7 @@ func (pm *PersistenceManager) SaveAllWithSync() error { } // LoadDatabase загружает базу данных с диска с применением WAL +// ИСПРАВЛЕНО: Добавлена проверка целостности снапшота и обработка torn write func (pm *PersistenceManager) LoadDatabase(dbName string) error { pm.isRestoring.Store(true) defer pm.isRestoring.Store(false) @@ -705,11 +741,12 @@ func (pm *PersistenceManager) LoadDatabase(dbName string) error { 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) + pm.tornWritesDetected.Add(1) + return fmt.Errorf("snapshot integrity check failed (possible torn write): %v", err) } // Создаём базу данных если не существует @@ -831,9 +868,11 @@ func (pm *PersistenceManager) LoadDatabase(dbName string) error { } else { for _, entry := range entries { if err := pm.applyWALEntry(db, entry); err != nil { + // ИСПРАВЛЕНО: Логируем, но продолжаем - возможно torn write if pm.logger != nil { - pm.logger.Error(fmt.Sprintf("Failed to apply WAL entry %d: %v", entry.Index, err)) + pm.logger.Error(fmt.Sprintf("Failed to apply WAL entry %d (possible torn write): %v", entry.Index, err)) } + pm.tornWritesDetected.Add(1) continue } appliedWalCount++ @@ -925,6 +964,7 @@ func (pm *PersistenceManager) applyWALEntry(db *Database, entry WALEntry) error } // verifySnapshot проверяет целостность снапшота +// ИСПРАВЛЕНО: Улучшена проверка целостности для обнаружения torn write func (pm *PersistenceManager) verifySnapshot(snapshot *DatabaseSnapshot) error { if snapshot.Checksum == "" { return fmt.Errorf("snapshot has no checksum") @@ -949,7 +989,8 @@ func (pm *PersistenceManager) verifySnapshot(snapshot *DatabaseSnapshot) error { checksum := hex.EncodeToString(hash[:]) if checksum != snapshot.Checksum { - return fmt.Errorf("checksum mismatch: expected %s, got %s", snapshot.Checksum[:8], checksum[:8]) + return fmt.Errorf("checksum mismatch: expected %s, got %s (possible torn write)", + snapshot.Checksum[:8], checksum[:8]) } return nil @@ -993,6 +1034,7 @@ func (pm *PersistenceManager) getSnapshotFilename(dbName string, version uint64) } // 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) @@ -1013,24 +1055,49 @@ func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapsh 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) + // ИСПРАВЛЕНО: Пытаемся загрузить снапшоты по порядку, пока не найдём валидный + for _, file := range files { + data, err := os.ReadFile(file) if err != nil { - return nil, err + if pm.logger != nil { + pm.logger.Warn(fmt.Sprintf("Failed to read snapshot %s: %v", file, err)) + } + continue } + + if pm.config.CompressEnabled && strings.HasSuffix(file, ".gz") { + data, err = pm.decompress(data) + if err != nil { + if pm.logger != nil { + pm.logger.Warn(fmt.Sprintf("Failed to decompress snapshot %s (possible torn write): %v", file, err)) + } + pm.tornWritesDetected.Add(1) + continue + } + } + + var snapshot DatabaseSnapshot + if err := json.Unmarshal(data, &snapshot); err != nil { + if pm.logger != nil { + pm.logger.Warn(fmt.Sprintf("Failed to unmarshal snapshot %s (possible torn write): %v", file, err)) + } + pm.tornWritesDetected.Add(1) + continue + } + + // ИСПРАВЛЕНО: Проверяем целостность + if err := pm.verifySnapshot(&snapshot); err != nil { + if pm.logger != nil { + pm.logger.Warn(fmt.Sprintf("Snapshot %s failed integrity check (possible torn write): %v", file, err)) + } + pm.tornWritesDetected.Add(1) + continue + } + + return &snapshot, nil } - var snapshot DatabaseSnapshot - if err := json.Unmarshal(data, &snapshot); err != nil { - return nil, err - } - - return &snapshot, nil + return nil, fmt.Errorf("no valid snapshot found (all may be corrupted)") } // compress сжимает данные с помощью gzip @@ -1059,7 +1126,7 @@ func (pm *PersistenceManager) decompress(data []byte) ([]byte, error) { var buf bytes.Buffer if _, err := buf.ReadFrom(reader); err != nil { - return nil, fmt.Errorf("failed to decompress data: %v", err) + return nil, fmt.Errorf("failed to decompress data (possible torn write): %v", err) } return buf.Bytes(), nil @@ -1113,6 +1180,7 @@ func (pm *PersistenceManager) GetLastCheckpointInfo() map[string]interface{} { "sync_attempts": pm.syncAttempts.Load(), "sync_failures": pm.syncFailures.Load(), "atomic_writes_count": pm.atomicWrites.Load(), + "torn_writes_detected": pm.tornWritesDetected.Load(), } if pm.recoveryInfo != nil { @@ -1133,3 +1201,8 @@ func (pm *PersistenceManager) GetRecoveryInfo() *CheckpointRecoveryInfo { func (pm *PersistenceManager) IsRestoring() bool { return pm.isRestoring.Load() } + +// GetTornWritesDetected возвращает количество обнаруженных torn write +func (pm *PersistenceManager) GetTornWritesDetected() uint64 { + return pm.tornWritesDetected.Load() +} \ No newline at end of file