Delete internal/storage/persistence.go
This commit is contained in:
@@ -1,466 +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"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"sort"
|
|
||||||
"strings"
|
|
||||||
"sync"
|
|
||||||
"sync/atomic"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
// 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"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// DefaultPersistenceConfig возвращает конфигурацию по умолчанию
|
|
||||||
func DefaultPersistenceConfig() *PersistenceConfig {
|
|
||||||
return &PersistenceConfig{
|
|
||||||
DataDir: "futriis_data",
|
|
||||||
CheckpointInterval: 5 * time.Minute,
|
|
||||||
MaxCheckpoints: 10,
|
|
||||||
CompressEnabled: true,
|
|
||||||
SyncWrites: 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"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// PersistenceManager управляет персистентным хранением
|
|
||||||
type PersistenceManager struct {
|
|
||||||
config *PersistenceConfig
|
|
||||||
storage *Storage
|
|
||||||
logger LoggerInterface
|
|
||||||
mu sync.RWMutex
|
|
||||||
stopChan chan struct{}
|
|
||||||
wg sync.WaitGroup
|
|
||||||
lastCheckpoint int64
|
|
||||||
checkpointID atomic.Uint64
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return pm
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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 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 {
|
|
||||||
pm.mu.Lock()
|
|
||||||
defer pm.mu.Unlock()
|
|
||||||
|
|
||||||
db, err := pm.storage.GetDatabase(dbName)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("database not found: %s", dbName)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Собираем данные всех коллекций
|
|
||||||
snapshot := &DatabaseSnapshot{
|
|
||||||
Name: dbName,
|
|
||||||
Collections: make(map[string]interface{}),
|
|
||||||
CreatedAt: time.Now().UnixMilli(),
|
|
||||||
Version: pm.checkpointID.Add(1),
|
|
||||||
}
|
|
||||||
|
|
||||||
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,
|
|
||||||
"version": doc.Version,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
snapshot.Collections[collName] = collData
|
|
||||||
}
|
|
||||||
|
|
||||||
// Сериализуем и сохраняем
|
|
||||||
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 := os.WriteFile(filename, data, 0644); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Синхронизируем с диском если нужно
|
|
||||||
if pm.config.SyncWrites {
|
|
||||||
if f, err := os.OpenFile(filename, os.O_RDWR, 0644); err == nil {
|
|
||||||
RealFsync(f)
|
|
||||||
f.Close()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if pm.logger != nil {
|
|
||||||
pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes)", dbName, snapshot.Version, len(data)))
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// SaveAll сохраняет все базы данных
|
|
||||||
func (pm *PersistenceManager) SaveAll() error {
|
|
||||||
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 загружает базу данных с диска
|
|
||||||
func (pm *PersistenceManager) LoadDatabase(dbName string) error {
|
|
||||||
pm.mu.Lock()
|
|
||||||
defer pm.mu.Unlock()
|
|
||||||
|
|
||||||
// Находим последний чекпоинт
|
|
||||||
snapshot, err := pm.findLatestSnapshot(dbName)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
if snapshot == nil {
|
|
||||||
return fmt.Errorf("no snapshot found for database: %s", dbName)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Восстанавливаем базу данных
|
|
||||||
if !pm.storage.ExistsDatabase(dbName) {
|
|
||||||
if err := pm.storage.CreateDatabase(dbName); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
db, err := pm.storage.GetDatabase(dbName)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Восстанавливаем коллекции
|
|
||||||
for collName, collDataRaw := range snapshot.Collections {
|
|
||||||
collData, ok := collDataRaw.([]interface{})
|
|
||||||
if !ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := db.GetCollection(collName); err != nil {
|
|
||||||
if err := db.CreateCollection(collName); err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
coll, err := db.GetCollection(collName)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
|
|
||||||
if updatedAt, ok := docMap["updated_at"].(float64); ok {
|
|
||||||
doc.UpdatedAt = int64(updatedAt)
|
|
||||||
}
|
|
||||||
|
|
||||||
if version, ok := docMap["version"].(float64); ok {
|
|
||||||
doc.Version = uint64(version)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := coll.Insert(doc); err != nil {
|
|
||||||
if pm.logger != nil {
|
|
||||||
pm.logger.Error(fmt.Sprintf("Failed to restore document %s: %v", docID, err))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if pm.logger != nil {
|
|
||||||
pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d)", dbName, snapshot.Version))
|
|
||||||
}
|
|
||||||
|
|
||||||
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)
|
|
||||||
// Извлекаем имя базы данных из имени файла
|
|
||||||
// Формат: snapshot_{dbname}_{version}.json.gz
|
|
||||||
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))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user