Update internal/storage/persistence.go
This commit is contained in:
gvsafronov committed
1 parent
3777ac7edc
commit
384553da25
1 file changed
+87
-14
@@ -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)
|
||||
@@ -289,6 +314,7 @@ type CheckpointRecoveryInfo struct {
|
||||
}
|
||||
|
||||
// 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])
|
||||
// ИСПРАВЛЕНО: Пытаемся загрузить снапшоты по порядку, пока не найдём валидный
|
||||
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(files[0], ".gz") {
|
||||
if pm.config.CompressEnabled && strings.HasSuffix(file, ".gz") {
|
||||
data, err = pm.decompress(data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
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 {
|
||||
return nil, err
|
||||
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
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
Reference in new issue
Block a user