diff --git a/internal/storage/transaction.go b/internal/storage/transaction.go deleted file mode 100644 index 3c3ac3b..0000000 --- a/internal/storage/transaction.go +++ /dev/null @@ -1,2241 +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/transaction.go -// Назначение: Реализация транзакций с поддержкой MVCC (Multi-Version Concurrency Control) -// и WAL (Write-Ahead Log) без блокировок. Использует атомарные операции и версионирование. -// ОПТИМИЗИРОВАНО: пакетная запись WAL, буферизация, асинхронная обработка. -// УЛУЧШЕНИЯ: Segmented WAL, Parallel Recovery, Visibility Map, Version Pruning, -// Distributed Transactions, Deadlock Detection, Transaction Timeout, Savepoints, -// Асинхронное восстановление WAL, реальный fsync. - -package storage - -import ( - "bufio" - "encoding/binary" - "encoding/json" - "fmt" - "os" - "path/filepath" - "sort" - "sync" - "sync/atomic" - "time" -) - -// ========== Базовые типы ========== - -// TransactionID представляет уникальный идентификатор транзакции -type TransactionID uint64 - -// TransactionState представляет состояние транзакции -type TransactionState int32 - -const ( - TransactionActive TransactionState = iota - TransactionCommitted - TransactionAborted -) - -// TransactionRecord представляет запись в WAL -type TransactionRecord struct { - ID TransactionID `json:"id"` - State TransactionState `json:"state"` - Timestamp int64 `json:"timestamp"` - Operations []Operation `json:"operations"` -} - -// Operation представляет одну операцию в транзакции -type Operation struct { - Type string `json:"type"` // "insert", "update", "delete" - Database string `json:"database"` - Collection string `json:"collection"` - DocumentID string `json:"document_id"` - Data map[string]interface{} `json:"data"` - Version uint64 `json:"version"` - OldData map[string]interface{} `json:"old_data"` // для отката -} - -// DocumentVersion представляет версию документа для MVCC -type DocumentVersion struct { - Document *Document `json:"document"` - Timestamp int64 `json:"timestamp"` - TxID TransactionID `json:"tx_id"` -} - -// TransactionInfo представляет информацию о транзакции для API -type TransactionInfo struct { - ID string `json:"id"` - Status string `json:"status"` - StartTime int64 `json:"start_time"` - OperationCount int `json:"operation_count"` - Operations []OperationInfo `json:"operations,omitempty"` -} - -// OperationInfo представляет информацию об операции для API -type OperationInfo struct { - Type string `json:"type"` - Database string `json:"database"` - Collection string `json:"collection"` - DocumentID string `json:"document_id"` -} - -// WALRecord представляет запись в WAL файле -type WALRecord struct { - CRC uint32 `json:"crc"` - Length uint32 `json:"length"` - Type byte `json:"type"` // 1=Transaction, 2=Checkpoint - Data []byte `json:"data"` - Timestamp int64 `json:"timestamp"` - LSN uint64 `json:"lsn"` // Log Sequence Number -} - -// LoggerInterface определяет интерфейс для логирования -type LoggerInterface interface { - Debug(msg string) - Info(msg string) - Error(msg string) - Warn(msg string) -} - -// ========== Константы для улучшений ========== - -const ( - // WAL Segment - WALSegmentSize = 64 * 1024 * 1024 // 64MB per segment - WALSegmentPrefix = "wal_segment_" - WALIndexPrefix = "wal_index_" - - // MVCC - VisibilityMapSize = 1024 * 1024 - VersionPruneInterval = 5 * time.Minute - MaxVersionsPerDoc = 100 - VersionRetentionDays = 7 - - // Transactions - DefaultTxTimeout = 30 * time.Second - DeadlockCheckInterval = 1 * time.Second - MaxSavepointsPerTx = 100 - - // Distributed Transactions - TwoPhaseCommitTimeout = 10 * time.Second - - // Асинхронное восстановление WAL - AsyncRecoveryBufferSize = 10000 - AsyncRecoveryWorkers = 4 - AsyncRecoveryTimeout = 30 * time.Second - - // Fsync - FsyncMaxRetries = 3 - FsyncRetryDelay = 100 * time.Millisecond -) - -// ========== CRC32 таблица ========== - -var crc32Table = [256]uint32{ - 0x00000000, 0x77073096, 0xee0e612c, 0x990951ba, 0x076dc419, 0x706af48f, - 0xe963a535, 0x9e6495a3, 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988, - 0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91, 0x1db71064, 0x6ab020f2, - 0xf3b97148, 0x84be41de, 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7, - 0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec, 0x14015c4f, 0x63066cd9, - 0xfa0f3d63, 0x8d080df5, 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172, - 0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b, 0x35b5a8fa, 0x42b2986c, - 0xdbbbc9d6, 0xacbcf940, 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59, - 0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116, 0x21b4f4b5, 0x56b3c423, - 0xcfba9599, 0xb8bda50f, 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924, - 0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d, 0x76dc4190, 0x01db7106, - 0x98d220bc, 0xefd5102a, 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433, - 0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818, 0x7f6a0dbb, 0x086d3d2d, - 0x91646c97, 0xe6635c01, 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e, - 0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457, 0x65b0d9c6, 0x12b7e950, - 0x8bbeb8ea, 0xfcb9887c, 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65, - 0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2, 0x4adfa541, 0x3dd895d7, - 0xa4d1c46d, 0xd3d6f4fb, 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0, - 0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9, 0x5005713c, 0x270241aa, - 0xbe0b1010, 0xc90c2086, 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f, - 0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4, 0x59b33d17, 0x2eb40d81, - 0xb7bd5c3b, 0xc0ba6cad, 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a, - 0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683, 0xe3630b12, 0x94643b84, - 0x0d6d6a3e, 0x7a6a5aa8, 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1, - 0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe, 0xf762575d, 0x806567cb, - 0x196c3671, 0x6e6b06e7, 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc, - 0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5, 0xd6d6a3e8, 0xa1d1937e, - 0x38d8c2c4, 0x4fdff252, 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b, - 0xd80d2bda, 0xaf0a1a4c, 0x36034af6, 0x41047a60, 0xdf60efc3, 0xa867df55, - 0x316e8eef, 0x4669be79, 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236, - 0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f, 0xc5ba3bbe, 0xb2bd0b28, - 0x2bb45a92, 0x5cb36a04, 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d, - 0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a, 0x9c0906a9, 0xeb0e363f, - 0x72076785, 0x05005713, 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38, - 0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21, 0x86d3d2d4, 0xf1d4e242, - 0x68ddb3f8, 0x1fda836e, 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777, - 0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c, 0x8f659eff, 0xf862ae69, - 0x616bffd3, 0x166ccf45, 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2, - 0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db, 0xaed16a4a, 0xd9d65adc, - 0x40df0b66, 0x37d83bf0, 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9, - 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6, 0xbad03605, 0xcdd70693, - 0x54de5729, 0x23d967bf, 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94, - 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d, -} - -// crc32 вычисляет CRC32 хеш -func crc32(data []byte) uint32 { - crc := uint32(0xFFFFFFFF) - for _, b := range data { - crc = (crc >> 8) ^ crc32Table[(crc^uint32(b))&0xFF] - } - return crc ^ 0xFFFFFFFF -} - -// ========== WALManager (оригинальный) ========== - -// WALManager управляет Write-Ahead Log с оптимизациями для высокой нагрузки -type WALManager struct { - mu sync.RWMutex - file *os.File - writer *bufio.Writer - path string - currentLSN uint64 - lastSync time.Time - syncInterval time.Duration - bufferSize int - closed bool - writeChan chan *WALRecord - stopChan chan struct{} - wg sync.WaitGroup - batchSize int -} - -// NewWALManager создаёт новый WAL менеджер с оптимизациями -func NewWALManager(path string) (*WALManager, error) { - dir := filepath.Dir(path) - if err := os.MkdirAll(dir, 0755); err != nil { - return nil, fmt.Errorf("failed to create WAL directory: %v", err) - } - - file, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0644) - if err != nil { - return nil, fmt.Errorf("failed to open WAL file: %v", err) - } - - var currentLSN uint64 = 1 - stat, err := file.Stat() - if err == nil && stat.Size() > 0 { - currentLSN = uint64(stat.Size()) / 100 - if currentLSN < 1 { - currentLSN = 1 - } - } - - wm := &WALManager{ - file: file, - writer: bufio.NewWriterSize(file, 64*1024), - path: path, - currentLSN: currentLSN, - lastSync: time.Now(), - syncInterval: 5 * time.Second, - bufferSize: 64 * 1024, - writeChan: make(chan *WALRecord, 10000), - stopChan: make(chan struct{}), - batchSize: 100, - } - - wm.wg.Add(1) - go wm.writerLoop() - - return wm, nil -} - -// writerLoop асинхронно записывает записи в WAL пакетами -func (wm *WALManager) writerLoop() { - defer wm.wg.Done() - - batch := make([]*WALRecord, 0, wm.batchSize) - ticker := time.NewTicker(wm.syncInterval) - defer ticker.Stop() - - for { - select { - case record, ok := <-wm.writeChan: - if !ok { - if len(batch) > 0 { - wm.flushBatch(batch) - } - return - } - batch = append(batch, record) - if len(batch) >= wm.batchSize { - wm.flushBatch(batch) - batch = batch[:0] - } - - case <-ticker.C: - if len(batch) > 0 { - wm.flushBatch(batch) - batch = batch[:0] - } - if time.Since(wm.lastSync) >= wm.syncInterval { - wm.sync() - } - - case <-wm.stopChan: - if len(batch) > 0 { - wm.flushBatch(batch) - } - wm.sync() - return - } - } -} - -// flushBatch записывает пакет записей -func (wm *WALManager) flushBatch(batch []*WALRecord) { - wm.mu.Lock() - defer wm.mu.Unlock() - - for _, record := range batch { - data, err := json.Marshal(record) - if err != nil { - continue - } - - lenBuf := make([]byte, 4) - binary.BigEndian.PutUint32(lenBuf, uint32(len(data))) - if _, err := wm.writer.Write(lenBuf); err != nil { - continue - } - - if _, err := wm.writer.Write(data); err != nil { - continue - } - - record.LSN = wm.currentLSN - wm.currentLSN++ - } -} - -// sync синхронизирует буфер с диском с реальным fsync -func (wm *WALManager) sync() { - wm.mu.Lock() - defer wm.mu.Unlock() - - if err := wm.writer.Flush(); err == nil { - // Реальный fsync с повторными попытками - if err := RealFsyncWithRetry(wm.file, FsyncMaxRetries, FsyncRetryDelay); err == nil { - wm.lastSync = time.Now() - } - } -} - -// Write записывает запись в WAL -func (wm *WALManager) Write(record *WALRecord) error { - if wm.closed { - return fmt.Errorf("WAL is closed") - } - - record.Timestamp = time.Now().UnixMilli() - - data, err := json.Marshal(record.Data) - if err != nil { - return err - } - record.CRC = crc32(data) - - select { - case wm.writeChan <- record: - return nil - case <-time.After(100 * time.Millisecond): - return fmt.Errorf("WAL write timeout") - } -} - -// ReadAll читает все записи из WAL -func (wm *WALManager) ReadAll() ([]*WALRecord, error) { - wm.mu.RLock() - defer wm.mu.RUnlock() - - wm.writer.Flush() - - file, err := os.Open(wm.path) - if err != nil { - return nil, err - } - defer file.Close() - - records := make([]*WALRecord, 0) - reader := bufio.NewReader(file) - lenBuf := make([]byte, 4) - - for { - _, err := reader.Read(lenBuf) - if err != nil { - break - } - - recordLen := binary.BigEndian.Uint32(lenBuf) - recordData := make([]byte, recordLen) - _, err = reader.Read(recordData) - if err != nil { - break - } - - var record WALRecord - if err := json.Unmarshal(recordData, &record); err != nil { - continue - } - - data, _ := json.Marshal(record.Data) - if crc32(data) != record.CRC { - continue - } - - records = append(records, &record) - } - - return records, nil -} - -// Close закрывает WAL с финальной синхронизацией -func (wm *WALManager) Close() error { - wm.mu.Lock() - wm.closed = true - wm.mu.Unlock() - - close(wm.stopChan) - close(wm.writeChan) - wm.wg.Wait() - - wm.mu.Lock() - defer wm.mu.Unlock() - - if err := wm.writer.Flush(); err != nil { - return err - } - // Финальный fsync - RealFsync(wm.file) - return wm.file.Close() -} - -// ========== Segmented WAL Manager ========== - -// WALSegment представляет сегмент WAL файла -type WALSegment struct { - ID uint32 - File *os.File - Writer *bufio.Writer - Path string - StartLSN uint64 - EndLSN uint64 - Size int64 - mu sync.Mutex -} - -// WALIndexEntry представляет запись в индексе WAL -type WALIndexEntry struct { - LSN uint64 - SegmentID uint32 - Offset int64 - Length uint32 - Checksum uint32 -} - -// WALIndexManager управляет индексом для быстрого поиска LSN -type WALIndexManager struct { - index map[uint64]*WALIndexEntry - segments map[uint32]*WALSegment - mu sync.RWMutex - indexPath string -} - -// SegmentedWALManager расширяет WALManager с поддержкой сегментов -type SegmentedWALManager struct { - segmentsDir string - segments map[uint32]*WALSegment - currentSegment *WALSegment - currentSegmentID uint32 - index *WALIndexManager - mu sync.RWMutex - writeChan chan *WALRecord - stopChan chan struct{} - wg sync.WaitGroup - batchSize int - logger LoggerInterface - recoveryManager *AsyncRecoveryManager - recoveryComplete atomic.Bool -} - -// NewSegmentedWALManager создаёт новый сегментированный WAL менеджер -func NewSegmentedWALManager(segmentsDir string, logger LoggerInterface) (*SegmentedWALManager, error) { - if err := os.MkdirAll(segmentsDir, 0755); err != nil { - return nil, fmt.Errorf("failed to create segments dir: %v", err) - } - - wm := &SegmentedWALManager{ - segmentsDir: segmentsDir, - segments: make(map[uint32]*WALSegment), - index: &WALIndexManager{ - index: make(map[uint64]*WALIndexEntry), - segments: make(map[uint32]*WALSegment), - indexPath: filepath.Join(segmentsDir, WALIndexPrefix+"index.json"), - }, - writeChan: make(chan *WALRecord, 10000), - stopChan: make(chan struct{}), - batchSize: 100, - logger: logger, - } - - if err := wm.loadExistingSegments(); err != nil { - return nil, err - } - - if err := wm.index.load(); err != nil { - if logger != nil { - logger.Warn(fmt.Sprintf("Failed to load WAL index: %v", err)) - } - } - - if wm.currentSegment == nil { - if err := wm.rotateSegment(); err != nil { - return nil, err - } - } - - wm.wg.Add(1) - go wm.writerLoop() - - return wm, nil -} - -// loadExistingSegments загружает существующие сегменты WAL -func (wm *SegmentedWALManager) loadExistingSegments() error { - files, err := filepath.Glob(filepath.Join(wm.segmentsDir, WALSegmentPrefix+"*")) - if err != nil { - return err - } - - for _, filePath := range files { - var segmentID uint32 - if _, err := fmt.Sscanf(filepath.Base(filePath), WALSegmentPrefix+"%d.log", &segmentID); err != nil { - continue - } - - file, err := os.OpenFile(filePath, os.O_RDWR, 0644) - if err != nil { - continue - } - - stat, _ := file.Stat() - segment := &WALSegment{ - ID: segmentID, - File: file, - Writer: bufio.NewWriterSize(file, 64*1024), - Path: filePath, - Size: stat.Size(), - StartLSN: uint64(segmentID) * WALSegmentSize / 100, - } - wm.segments[segmentID] = segment - - if segmentID > wm.currentSegmentID { - wm.currentSegmentID = segmentID - wm.currentSegment = segment - } - } - return nil -} - -// rotateSegment создаёт новый сегмент WAL -func (wm *SegmentedWALManager) rotateSegment() error { - wm.mu.Lock() - defer wm.mu.Unlock() - - newSegmentID := wm.currentSegmentID + 1 - segmentPath := filepath.Join(wm.segmentsDir, fmt.Sprintf(WALSegmentPrefix+"%d.log", newSegmentID)) - - file, err := os.OpenFile(segmentPath, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0644) - if err != nil { - return fmt.Errorf("failed to create segment: %v", err) - } - - newSegment := &WALSegment{ - ID: newSegmentID, - File: file, - Writer: bufio.NewWriterSize(file, 64*1024), - Path: segmentPath, - StartLSN: wm.getCurrentLSN(), - } - - if wm.currentSegment != nil { - wm.currentSegment.Writer.Flush() - RealFsync(wm.currentSegment.File) - wm.currentSegment.File.Close() - } - - wm.currentSegment = newSegment - wm.currentSegmentID = newSegmentID - wm.segments[newSegmentID] = newSegment - - if wm.logger != nil { - wm.logger.Info(fmt.Sprintf("Created new WAL segment: %d", newSegmentID)) - } - - return nil -} - -// getCurrentLSN возвращает текущий LSN -func (wm *SegmentedWALManager) getCurrentLSN() uint64 { - wm.mu.RLock() - defer wm.mu.RUnlock() - if wm.currentSegment == nil { - return 1 - } - return wm.currentSegment.StartLSN + uint64(wm.currentSegment.Size/100) -} - -// Write записывает запись в WAL -func (wm *SegmentedWALManager) Write(record *WALRecord) error { - record.Timestamp = time.Now().UnixMilli() - wm.writeChan <- record - return nil -} - -// writerLoop асинхронно записывает записи -func (wm *SegmentedWALManager) writerLoop() { - defer wm.wg.Done() - - batch := make([]*WALRecord, 0, wm.batchSize) - ticker := time.NewTicker(5 * time.Second) - defer ticker.Stop() - - for { - select { - case record, ok := <-wm.writeChan: - if !ok { - wm.flushBatch(batch) - return - } - batch = append(batch, record) - if len(batch) >= wm.batchSize { - wm.flushBatch(batch) - batch = batch[:0] - } - - case <-ticker.C: - if len(batch) > 0 { - wm.flushBatch(batch) - batch = batch[:0] - } - - case <-wm.stopChan: - wm.flushBatch(batch) - return - } - } -} - -// flushBatch записывает пакет записей с реальным fsync -func (wm *SegmentedWALManager) flushBatch(batch []*WALRecord) { - wm.mu.Lock() - defer wm.mu.Unlock() - - for _, record := range batch { - if wm.currentSegment.Size >= WALSegmentSize { - wm.mu.Unlock() - wm.rotateSegment() - wm.mu.Lock() - } - - data, err := json.Marshal(record) - if err != nil { - continue - } - - lsnBytes := make([]byte, 8) - binary.BigEndian.PutUint64(lsnBytes, record.LSN) - crcData := append(lsnBytes, data...) - record.CRC = crc32(crcData) - - lenBuf := make([]byte, 4) - binary.BigEndian.PutUint32(lenBuf, uint32(len(data))) - if _, err := wm.currentSegment.Writer.Write(lenBuf); err != nil { - continue - } - - if _, err := wm.currentSegment.Writer.Write(data); err != nil { - continue - } - - wm.index.addEntry(&WALIndexEntry{ - LSN: record.LSN, - SegmentID: wm.currentSegment.ID, - Offset: wm.currentSegment.Size, - Length: uint32(len(data)), - Checksum: record.CRC, - }) - - wm.currentSegment.Size += int64(4 + len(data)) - wm.currentSegment.EndLSN = record.LSN - } - - wm.currentSegment.Writer.Flush() - // Реальный fsync после записи пакета - RealFsync(wm.currentSegment.File) - wm.index.save() -} - -// ReadAll читает все записи из всех сегментов WAL -func (wm *SegmentedWALManager) ReadAll() ([]*WALRecord, error) { - wm.mu.RLock() - segments := make([]*WALSegment, 0, len(wm.segments)) - for _, seg := range wm.segments { - segments = append(segments, seg) - } - wm.mu.RUnlock() - - sort.Slice(segments, func(i, j int) bool { - return segments[i].ID < segments[j].ID - }) - - records := make([]*WALRecord, 0) - - for _, seg := range segments { - segRecords, err := wm.readSegmentRecords(seg) - if err != nil { - return nil, err - } - records = append(records, segRecords...) - } - - return records, nil -} - -// readSegmentRecords читает все записи из одного сегмента -func (wm *SegmentedWALManager) readSegmentRecords(seg *WALSegment) ([]*WALRecord, error) { - seg.mu.Lock() - defer seg.mu.Unlock() - - if seg.File == nil { - return nil, nil - } - - seg.Writer.Flush() - seg.File.Seek(0, 0) - - records := make([]*WALRecord, 0) - reader := bufio.NewReader(seg.File) - lenBuf := make([]byte, 4) - - for { - _, err := reader.Read(lenBuf) - if err != nil { - break - } - - recordLen := binary.BigEndian.Uint32(lenBuf) - recordData := make([]byte, recordLen) - _, err = reader.Read(recordData) - if err != nil { - break - } - - var record WALRecord - if err := json.Unmarshal(recordData, &record); err != nil { - continue - } - - lsnBytes := make([]byte, 8) - binary.BigEndian.PutUint64(lsnBytes, record.LSN) - crcData := append(lsnBytes, recordData...) - if crc32(crcData) != record.CRC { - continue - } - - records = append(records, &record) - } - - return records, nil -} - -// RecoverParallel восстанавливает транзакции параллельно -func (wm *SegmentedWALManager) RecoverParallel(callback func(*WALRecord) error) error { - wm.mu.RLock() - segments := make([]*WALSegment, 0, len(wm.segments)) - for _, seg := range wm.segments { - segments = append(segments, seg) - } - wm.mu.RUnlock() - - sort.Slice(segments, func(i, j int) bool { - return segments[i].ID < segments[j].ID - }) - - const numWorkers = 4 - recordChan := make(chan *WALRecord, 1000) - errChan := make(chan error, numWorkers) - var wg sync.WaitGroup - - for i := 0; i < numWorkers; i++ { - wg.Add(1) - go func() { - defer wg.Done() - for record := range recordChan { - if err := callback(record); err != nil { - select { - case errChan <- err: - default: - } - } - } - }() - } - - go func() { - defer close(recordChan) - for _, seg := range segments { - if err := wm.readSegment(seg, recordChan); err != nil { - select { - case errChan <- err: - default: - } - return - } - } - }() - - wg.Wait() - close(errChan) - - if len(errChan) > 0 { - return <-errChan - } - return nil -} - -// readSegment читает сегмент и отправляет записи в канал -func (wm *SegmentedWALManager) readSegment(seg *WALSegment, recordChan chan<- *WALRecord) error { - seg.mu.Lock() - defer seg.mu.Unlock() - - if seg.File == nil { - return nil - } - - seg.Writer.Flush() - seg.File.Seek(0, 0) - - reader := bufio.NewReader(seg.File) - lenBuf := make([]byte, 4) - - for { - _, err := reader.Read(lenBuf) - if err != nil { - break - } - - recordLen := binary.BigEndian.Uint32(lenBuf) - recordData := make([]byte, recordLen) - _, err = reader.Read(recordData) - if err != nil { - break - } - - var record WALRecord - if err := json.Unmarshal(recordData, &record); err != nil { - continue - } - - lsnBytes := make([]byte, 8) - binary.BigEndian.PutUint64(lsnBytes, record.LSN) - crcData := append(lsnBytes, recordData...) - if crc32(crcData) != record.CRC { - continue - } - - select { - case recordChan <- &record: - default: - } - } - - return nil -} - -// Close закрывает SegmentedWALManager -func (wm *SegmentedWALManager) Close() error { - close(wm.stopChan) - close(wm.writeChan) - wm.wg.Wait() - - wm.mu.Lock() - defer wm.mu.Unlock() - - if wm.currentSegment != nil { - wm.currentSegment.Writer.Flush() - RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay) - wm.currentSegment.File.Close() - } - - wm.index.save() - return nil -} - -// addEntry добавляет запись в индекс -func (im *WALIndexManager) addEntry(entry *WALIndexEntry) { - im.mu.Lock() - defer im.mu.Unlock() - im.index[entry.LSN] = entry -} - -// FindByLSN ищет запись по LSN -func (im *WALIndexManager) FindByLSN(lsn uint64) *WALIndexEntry { - im.mu.RLock() - defer im.mu.RUnlock() - return im.index[lsn] -} - -// save сохраняет индекс на диск -func (im *WALIndexManager) save() error { - im.mu.RLock() - defer im.mu.RUnlock() - - data, err := json.Marshal(im.index) - if err != nil { - return err - } - return os.WriteFile(im.indexPath, data, 0644) -} - -// load загружает индекс с диска -func (im *WALIndexManager) load() error { - data, err := os.ReadFile(im.indexPath) - if err != nil { - if os.IsNotExist(err) { - return nil - } - return err - } - return json.Unmarshal(data, &im.index) -} - -// ========== Async Recovery Manager ========== - -// AsyncRecoveryManager управляет асинхронным восстановлением из WAL -type AsyncRecoveryManager struct { - recordChan chan *WALRecord - errChan chan error - doneChan chan struct{} - wg sync.WaitGroup - callback func(*WALRecord) error - mu sync.RWMutex - isRunning bool - recoveredCnt atomic.Uint64 - errorCnt atomic.Uint64 - startTime time.Time -} - -// NewAsyncRecoveryManager создаёт менеджер асинхронного восстановления -func NewAsyncRecoveryManager(callback func(*WALRecord) error, workers int) *AsyncRecoveryManager { - arm := &AsyncRecoveryManager{ - recordChan: make(chan *WALRecord, AsyncRecoveryBufferSize), - errChan: make(chan error, workers), - doneChan: make(chan struct{}), - callback: callback, - startTime: time.Now(), - isRunning: true, - } - - for i := 0; i < workers; i++ { - arm.wg.Add(1) - go arm.worker() - } - - go arm.errorMonitor() - - return arm -} - -// worker обрабатывает записи из канала -func (arm *AsyncRecoveryManager) worker() { - defer arm.wg.Done() - - for record := range arm.recordChan { - if err := arm.callback(record); err != nil { - select { - case arm.errChan <- err: - default: - } - arm.errorCnt.Add(1) - } else { - arm.recoveredCnt.Add(1) - } - } -} - -// errorMonitor отслеживает ошибки и может остановить процесс -func (arm *AsyncRecoveryManager) errorMonitor() { - criticalErrors := 0 - for range arm.errChan { - criticalErrors++ - if criticalErrors > 10 { - arm.Stop() - return - } - } -} - -// Push добавляет запись для восстановления -func (arm *AsyncRecoveryManager) Push(record *WALRecord) bool { - arm.mu.RLock() - if !arm.isRunning { - arm.mu.RUnlock() - return false - } - arm.mu.RUnlock() - - select { - case arm.recordChan <- record: - return true - case <-time.After(100 * time.Millisecond): - return false - } -} - -// Wait ожидает завершения всех воркеров -func (arm *AsyncRecoveryManager) Wait() { - close(arm.recordChan) - arm.wg.Wait() - close(arm.doneChan) -} - -// WaitWithTimeout ожидает завершения с таймаутом -func (arm *AsyncRecoveryManager) WaitWithTimeout(timeout time.Duration) bool { - done := make(chan struct{}) - go func() { - arm.Wait() - close(done) - }() - - select { - case <-done: - return true - case <-time.After(timeout): - return false - } -} - -// Stop останавливает восстановление -func (arm *AsyncRecoveryManager) Stop() { - arm.mu.Lock() - if !arm.isRunning { - arm.mu.Unlock() - return - } - arm.isRunning = false - arm.mu.Unlock() - - close(arm.recordChan) -} - -// GetStats возвращает статистику восстановления -func (arm *AsyncRecoveryManager) GetStats() map[string]interface{} { - return map[string]interface{}{ - "recovered": arm.recoveredCnt.Load(), - "errors": arm.errorCnt.Load(), - "is_running": arm.isRunning, - "elapsed_ms": time.Since(arm.startTime).Milliseconds(), - } -} - -// IsComplete возвращает статус завершения -func (arm *AsyncRecoveryManager) IsComplete() bool { - return !arm.isRunning && arm.recoveredCnt.Load() > 0 -} - -// ========== MVCC Improvements ========== - -// VisibilityMapEntry представляет запись в карте видимости -type VisibilityMapEntry struct { - DocID string - VisibleFrom uint64 - VisibleTo uint64 - IsVisible bool - LastAccess int64 -} - -// VisibilityMap управляет битовой картой видимости версий -type VisibilityMap struct { - entries sync.Map - maxSize int - hitCount atomic.Uint64 - missCount atomic.Uint64 -} - -// NewVisibilityMap создаёт новую карту видимости -func NewVisibilityMap(maxSize int) *VisibilityMap { - if maxSize <= 0 { - maxSize = VisibilityMapSize - } - return &VisibilityMap{maxSize: maxSize} -} - -// MarkVisible отмечает версию как видимую -func (vm *VisibilityMap) MarkVisible(docID string, version uint64, visible bool) { - key := fmt.Sprintf("%s@%d", docID, version) - vm.entries.Store(key, &VisibilityMapEntry{ - DocID: docID, - VisibleFrom: version, - VisibleTo: version, - IsVisible: visible, - LastAccess: time.Now().Unix(), - }) -} - -// IsVisible проверяет видимость версии -func (vm *VisibilityMap) IsVisible(docID string, version uint64) bool { - key := fmt.Sprintf("%s@%d", docID, version) - _, ok := vm.entries.Load(key) - if !ok { - vm.missCount.Add(1) - return false - } - vm.hitCount.Add(1) - return true -} - -// GetStats возвращает статистику -func (vm *VisibilityMap) GetStats() map[string]interface{} { - return map[string]interface{}{ - "hits": vm.hitCount.Load(), - "misses": vm.missCount.Load(), - } -} - -// VersionPrunePolicy определяет политику очистки версий -type VersionPrunePolicy struct { - MaxVersions int - RetentionDays int - MinVersionAge time.Duration - PruneInterval time.Duration -} - -// VersionPruner управляет очисткой старых версий -type VersionPruner struct { - policy *VersionPrunePolicy - txMgr *TransactionManager - logger LoggerInterface - stopChan chan struct{} - wg sync.WaitGroup -} - -// NewVersionPruner создаёт новый очиститель версий -func NewVersionPruner(policy *VersionPrunePolicy, txMgr *TransactionManager, logger LoggerInterface) *VersionPruner { - if policy == nil { - policy = &VersionPrunePolicy{ - MaxVersions: MaxVersionsPerDoc, - RetentionDays: VersionRetentionDays, - MinVersionAge: time.Hour, - PruneInterval: VersionPruneInterval, - } - } - return &VersionPruner{ - policy: policy, - txMgr: txMgr, - logger: logger, - stopChan: make(chan struct{}), - } -} - -// Start запускает фоновую очистку -func (vp *VersionPruner) Start() { - vp.wg.Add(1) - go vp.pruneLoop() -} - -// Stop останавливает очистку -func (vp *VersionPruner) Stop() { - close(vp.stopChan) - vp.wg.Wait() -} - -// pruneLoop периодически очищает старые версии -func (vp *VersionPruner) pruneLoop() { - defer vp.wg.Done() - ticker := time.NewTicker(vp.policy.PruneInterval) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - vp.prune() - case <-vp.stopChan: - return - } - } -} - -// prune выполняет очистку старых версий -func (vp *VersionPruner) prune() { - cutoffTime := time.Now().AddDate(0, 0, -vp.policy.RetentionDays).UnixMilli() - - vp.txMgr.documentVersions.Range(func(key, value interface{}) bool { - versions := value.([]*DocumentVersion) - if len(versions) <= vp.policy.MaxVersions { - return true - } - - newVersions := make([]*DocumentVersion, 0, vp.policy.MaxVersions) - for _, v := range versions { - if v.Timestamp >= cutoffTime && len(newVersions) < vp.policy.MaxVersions { - newVersions = append(newVersions, v) - } - } - - if len(newVersions) < len(versions) { - vp.txMgr.documentVersions.Store(key, newVersions) - if vp.logger != nil { - vp.logger.Debug(fmt.Sprintf("Pruned %d versions for doc %v", len(versions)-len(newVersions), key)) - } - } - return true - }) -} - -// ReadTimestampCache кэширует версии документов по timestamp -type ReadTimestampCache struct { - cache sync.Map - maxSize int - ttl time.Duration - hits atomic.Uint64 - misses atomic.Uint64 -} - -type cachedEntry struct { - doc *Document - cachedAt time.Time -} - -// NewReadTimestampCache создаёт новый кэш -func NewReadTimestampCache(maxSize int, ttl time.Duration) *ReadTimestampCache { - if maxSize <= 0 { - maxSize = 10000 - } - if ttl <= 0 { - ttl = 5 * time.Minute - } - return &ReadTimestampCache{maxSize: maxSize, ttl: ttl} -} - -// Get возвращает кэшированную версию документа -func (rtc *ReadTimestampCache) Get(docID string, timestamp int64) *Document { - key := fmt.Sprintf("%s@%d", docID, timestamp) - val, ok := rtc.cache.Load(key) - if !ok { - rtc.misses.Add(1) - return nil - } - entry := val.(*cachedEntry) - if time.Since(entry.cachedAt) > rtc.ttl { - rtc.cache.Delete(key) - rtc.misses.Add(1) - return nil - } - rtc.hits.Add(1) - return entry.doc -} - -// Set добавляет версию в кэш -func (rtc *ReadTimestampCache) Set(docID string, timestamp int64, doc *Document) { - key := fmt.Sprintf("%s@%d", docID, timestamp) - rtc.cache.Store(key, &cachedEntry{doc: doc, cachedAt: time.Now()}) -} - -// GetStats возвращает статистику кэша -func (rtc *ReadTimestampCache) GetStats() map[string]interface{} { - return map[string]interface{}{ - "hits": rtc.hits.Load(), - "misses": rtc.misses.Load(), - } -} - -// ========== Distributed Transaction Coordinator ========== - -// DistributedTxInfo содержит информацию о распределённой транзакции -type DistributedTxInfo struct { - TxID TransactionID - Nodes []string - Prepared map[string]bool - Committed map[string]bool - Status TxState - StartTime int64 - Timeout time.Duration -} - -// TxState представляет состояние транзакции -type TxState int32 - -const ( - TxActive TxState = iota - TxPrepared - TxCommitted - TxAborted - TxTimeout -) - -// DistributedTransactionCoordinator координирует распределённые транзакции -type DistributedTransactionCoordinator struct { - pendingTxs sync.Map - timeout time.Duration - mu sync.RWMutex -} - -// NewDistributedTransactionCoordinator создаёт новый координатор -func NewDistributedTransactionCoordinator(timeout time.Duration) *DistributedTransactionCoordinator { - if timeout <= 0 { - timeout = TwoPhaseCommitTimeout - } - return &DistributedTransactionCoordinator{timeout: timeout} -} - -// Prepare подготавливает распределённую транзакцию -func (dtc *DistributedTransactionCoordinator) Prepare(txID TransactionID, nodes []string) error { - info := &DistributedTxInfo{ - TxID: txID, - Nodes: nodes, - Prepared: make(map[string]bool), - Committed: make(map[string]bool), - Status: TxActive, - StartTime: time.Now().UnixMilli(), - Timeout: dtc.timeout, - } - dtc.pendingTxs.Store(txID, info) - return nil -} - -// VoteCommit получает голоса за коммит от узлов -func (dtc *DistributedTransactionCoordinator) VoteCommit(txID TransactionID, node string) error { - val, ok := dtc.pendingTxs.Load(txID) - if !ok { - return fmt.Errorf("transaction not found: %d", txID) - } - info := val.(*DistributedTxInfo) - - dtc.mu.Lock() - info.Prepared[node] = true - dtc.mu.Unlock() - - if len(info.Prepared) == len(info.Nodes) { - return dtc.Commit(txID) - } - return nil -} - -// Commit коммитит распределённую транзакцию -func (dtc *DistributedTransactionCoordinator) Commit(txID TransactionID) error { - val, ok := dtc.pendingTxs.Load(txID) - if !ok { - return fmt.Errorf("transaction not found: %d", txID) - } - info := val.(*DistributedTxInfo) - info.Status = TxCommitted - return nil -} - -// Abort откатывает распределённую транзакцию -func (dtc *DistributedTransactionCoordinator) Abort(txID TransactionID) error { - dtc.pendingTxs.Delete(txID) - return nil -} - -// ========== Deadlock Detector ========== - -// DeadlockDetector обнаруживает циклические зависимости между транзакциями -type DeadlockDetector struct { - waitForGraph sync.Map - checkInterval time.Duration - timeout time.Duration - mu sync.RWMutex - stopChan chan struct{} - wg sync.WaitGroup - logger LoggerInterface -} - -// NewDeadlockDetector создаёт новый детектор дедлоков -func NewDeadlockDetector(checkInterval, timeout time.Duration) *DeadlockDetector { - if checkInterval <= 0 { - checkInterval = DeadlockCheckInterval - } - if timeout <= 0 { - timeout = DefaultTxTimeout - } - d := &DeadlockDetector{ - checkInterval: checkInterval, - timeout: timeout, - stopChan: make(chan struct{}), - } - d.wg.Add(1) - go d.detectLoop() - return d -} - -// SetLogger устанавливает логгер для детектора дедлоков -func (dd *DeadlockDetector) SetLogger(logger LoggerInterface) { - dd.logger = logger -} - -// Stop останавливает детектор -func (dd *DeadlockDetector) Stop() { - close(dd.stopChan) - dd.wg.Wait() -} - -// AddWaiting добавляет ожидающую транзакцию -func (dd *DeadlockDetector) AddWaiting(waiting, waitingFor TransactionID) { - var list []TransactionID - if val, ok := dd.waitForGraph.Load(waiting); ok { - list = val.([]TransactionID) - } - list = append(list, waitingFor) - dd.waitForGraph.Store(waiting, list) -} - -// RemoveWaiting удаляет ожидающую транзакцию -func (dd *DeadlockDetector) RemoveWaiting(txID TransactionID) { - dd.waitForGraph.Delete(txID) -} - -// detectLoop обнаруживает циклы в графе ожидания -func (dd *DeadlockDetector) detectLoop() { - defer dd.wg.Done() - ticker := time.NewTicker(dd.checkInterval) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - dd.detect() - case <-dd.stopChan: - return - } - } -} - -// detect выполняет обнаружение дедлоков -func (dd *DeadlockDetector) detect() { - visited := make(map[TransactionID]bool) - stack := make(map[TransactionID]bool) - - var dfs func(txID TransactionID) bool - dfs = func(txID TransactionID) bool { - visited[txID] = true - stack[txID] = true - - val, ok := dd.waitForGraph.Load(txID) - if ok { - for _, next := range val.([]TransactionID) { - if !visited[next] { - if dfs(next) { - return true - } - } else if stack[next] { - dd.resolveDeadlock(txID, next) - return true - } - } - } - stack[txID] = false - return false - } - - dd.waitForGraph.Range(func(key, value interface{}) bool { - txID := key.(TransactionID) - if !visited[txID] { - dfs(txID) - } - return true - }) -} - -// resolveDeadlock разрешает дедлок, откатывая одну из транзакций -func (dd *DeadlockDetector) resolveDeadlock(txID1, txID2 TransactionID) { - if globalTxManager != nil { - if val, ok := globalTxManager.activeTransactions.Load(txID1); ok { - tx := val.(*Transaction) - tx.State.Store(int32(TransactionAborted)) - globalTxManager.activeTransactions.Delete(txID1) - if dd.logger != nil { - dd.logger.Warn(fmt.Sprintf("Deadlock resolved: aborted transaction %d due to conflict with %d", txID1, txID2)) - } - } - } -} - -// ========== Transaction With Timeout and Savepoints ========== - -// Savepoint представляет точку сохранения в транзакции -type Savepoint struct { - ID string - Timestamp int64 - Operations []Operation - Version uint64 -} - -// TransactionWithTimeout расширяет Transaction с поддержкой таймаута и savepoints -type TransactionWithTimeout struct { - *Transaction - Timeout time.Duration - StartTime time.Time - Timer *time.Timer - Savepoints []*Savepoint - mu sync.RWMutex -} - -// NewTransactionWithTimeout создаёт транзакцию с таймаутом -func NewTransactionWithTimeout(timeout time.Duration) *TransactionWithTimeout { - tx := BeginTransaction() - if tx == nil { - return nil - } - - txw := &TransactionWithTimeout{ - Transaction: tx, - Timeout: timeout, - StartTime: time.Now(), - Savepoints: make([]*Savepoint, 0, MaxSavepointsPerTx), - } - - txw.Timer = time.AfterFunc(timeout, func() { - txw.State.Store(int32(TransactionAborted)) - if globalTxManager != nil { - globalTxManager.activeTransactions.Delete(txw.ID) - } - AuditLog("TIMEOUT", "TRANSACTION", fmt.Sprintf("%d", txw.ID), map[string]interface{}{ - "timeout_ms": timeout.Milliseconds(), - }) - }) - - return txw -} - -// CreateSavepoint создаёт точку сохранения -func (txw *TransactionWithTimeout) CreateSavepoint(id string) error { - txw.mu.Lock() - defer txw.mu.Unlock() - - if len(txw.Savepoints) >= MaxSavepointsPerTx { - return fmt.Errorf("too many savepoints, max %d", MaxSavepointsPerTx) - } - - opsCopy := make([]Operation, len(txw.Operations)) - copy(opsCopy, txw.Operations) - - savepoint := &Savepoint{ - ID: id, - Timestamp: time.Now().UnixMilli(), - Operations: opsCopy, - Version: txw.Version, - } - txw.Savepoints = append(txw.Savepoints, savepoint) - - AuditLog("SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d.%s", txw.ID, id), map[string]interface{}{ - "operation_count": len(opsCopy), - }) - - return nil -} - -// RollbackToSavepoint откатывает до точки сохранения -func (txw *TransactionWithTimeout) RollbackToSavepoint(id string) error { - txw.mu.Lock() - defer txw.mu.Unlock() - - var targetSavepoint *Savepoint - var savepointIndex int - - for i, sp := range txw.Savepoints { - if sp.ID == id { - targetSavepoint = sp - savepointIndex = i - break - } - } - - if targetSavepoint == nil { - return fmt.Errorf("savepoint not found: %s", id) - } - - txw.Operations = make([]Operation, len(targetSavepoint.Operations)) - copy(txw.Operations, targetSavepoint.Operations) - - txw.Savepoints = txw.Savepoints[:savepointIndex+1] - txw.Version = targetSavepoint.Version - - AuditLog("ROLLBACK_SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d.%s", txw.ID, id), map[string]interface{}{ - "rolled_back_to": targetSavepoint.Timestamp, - }) - - return nil -} - -// ReleaseSavepoint освобождает точку сохранения -func (txw *TransactionWithTimeout) ReleaseSavepoint(id string) error { - txw.mu.Lock() - defer txw.mu.Unlock() - - for i, sp := range txw.Savepoints { - if sp.ID == id { - txw.Savepoints = append(txw.Savepoints[:i], txw.Savepoints[i+1:]...) - return nil - } - } - return fmt.Errorf("savepoint not found: %s", id) -} - -// ========== Transaction (основная структура) ========== - -// Transaction представляет одну транзакцию -type Transaction struct { - ID TransactionID - State atomic.Int32 - Operations []Operation - StartTime int64 - Version uint64 - mu sync.RWMutex -} - -// TransactionManager управляет транзакциями с поддержкой MVCC -type TransactionManager struct { - activeTransactions sync.Map // map[TransactionID]*Transaction - nextTxID atomic.Uint64 - wal *SegmentedWALManager - logger LoggerInterface - mu sync.RWMutex - walPath string - checkpointInterval int64 - lastCheckpoint int64 - checkpointFile *os.File - documentVersions sync.Map // map[string][]*DocumentVersion - maxVersions int - visibilityMap *VisibilityMap - readCache *ReadTimestampCache - distCoord *DistributedTransactionCoordinator - deadlockDetector *DeadlockDetector - recoveryManager *AsyncRecoveryManager - recoveryComplete atomic.Bool -} - -var ( - globalTxManager *TransactionManager - txManagerOnce sync.Once - currentTx atomic.Value // *Transaction - globalStorage *Storage -) - -// InitTransactionManager инициализирует глобальный менеджер транзакций -func InitTransactionManager(walPath string) error { - return InitTransactionManagerWithConfig(walPath, nil) -} - -// InitTransactionManagerWithConfig инициализирует менеджер транзакций с конфигурацией -func InitTransactionManagerWithConfig(walPath string, config map[string]interface{}) error { - var err error - txManagerOnce.Do(func() { - maxVersions := 10 - if config != nil { - if v, ok := config["max_versions"].(int); ok && v > 0 { - maxVersions = v - } - } - - globalTxManager = &TransactionManager{ - nextTxID: atomic.Uint64{}, - walPath: walPath, - checkpointInterval: 300, - lastCheckpoint: time.Now().Unix(), - maxVersions: maxVersions, - visibilityMap: NewVisibilityMap(VisibilityMapSize), - readCache: NewReadTimestampCache(10000, 5*time.Minute), - distCoord: NewDistributedTransactionCoordinator(TwoPhaseCommitTimeout), - deadlockDetector: NewDeadlockDetector(DeadlockCheckInterval, DefaultTxTimeout), - } - globalTxManager.nextTxID.Store(1) - - var walErr error - globalTxManager.wal, walErr = NewSegmentedWALManager(filepath.Dir(walPath), nil) - if walErr != nil { - err = walErr - return - } - - globalTxManager.startAsyncRecovery() - - go globalTxManager.checkpointLoop() - go globalTxManager.versionCleanupLoop() - }) - return err -} - -// startAsyncRecovery запускает асинхронное восстановление из WAL -func (tm *TransactionManager) startAsyncRecovery() { - if tm.wal == nil { - tm.recoveryComplete.Store(true) - return - } - - if tm.logger != nil { - tm.logger.Info("Starting asynchronous WAL recovery...") - } - - records, err := tm.wal.ReadAll() - if err != nil { - if tm.logger != nil { - tm.logger.Error(fmt.Sprintf("Failed to read WAL: %v", err)) - } - tm.recoveryComplete.Store(true) - return - } - - if len(records) == 0 { - if tm.logger != nil { - tm.logger.Info("No records to recover") - } - tm.recoveryComplete.Store(true) - return - } - - tm.recoveryManager = NewAsyncRecoveryManager(func(record *WALRecord) error { - if record.Type == 1 { - var txRecord TransactionRecord - if err := json.Unmarshal(record.Data, &txRecord); err != nil { - return err - } - - if txRecord.State == TransactionCommitted { - for _, op := range txRecord.Operations { - if err := applyOperation(op); err != nil { - return err - } - } - } - } - return nil - }, AsyncRecoveryWorkers) - - for _, record := range records { - if !tm.recoveryManager.Push(record) { - if tm.logger != nil { - tm.logger.Warn("Recovery buffer full, some records may be delayed") - } - } - } - - go func() { - ticker := time.NewTicker(5 * time.Second) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - stats := tm.recoveryManager.GetStats() - if tm.logger != nil { - tm.logger.Debug(fmt.Sprintf("Recovery progress: %d/%d records recovered", - stats["recovered"], len(records))) - } - case <-tm.recoveryManager.doneChan: - stats := tm.recoveryManager.GetStats() - if tm.logger != nil { - tm.logger.Info(fmt.Sprintf("WAL recovery completed: %d records recovered, %d errors", - stats["recovered"], stats["errors"])) - } - tm.recoveryComplete.Store(true) - return - } - } - }() -} - -// IsRecoveryComplete возвращает статус восстановления -func (tm *TransactionManager) IsRecoveryComplete() bool { - return tm.recoveryComplete.Load() -} - -// GetRecoveryProgress возвращает прогресс восстановления -func (tm *TransactionManager) GetRecoveryProgress() map[string]interface{} { - if tm.recoveryManager == nil { - return map[string]interface{}{ - "is_recovering": false, - "recovered": 0, - "complete": true, - } - } - - stats := tm.recoveryManager.GetStats() - return map[string]interface{}{ - "is_recovering": !tm.recoveryComplete.Load(), - "recovered": stats["recovered"], - "complete": tm.recoveryComplete.Load(), - "elapsed_ms": stats["elapsed_ms"], - } -} - -// SetTransactionLogger устанавливает логгер для транзакций -func SetTransactionLogger(logger LoggerInterface) { - if globalTxManager != nil { - globalTxManager.logger = logger - if globalTxManager.wal != nil { - globalTxManager.wal.logger = logger - } - if globalTxManager.deadlockDetector != nil { - globalTxManager.deadlockDetector.SetLogger(logger) - } - } -} - -// BeginTransaction начинает новую транзакцию -func BeginTransaction() *Transaction { - if globalTxManager == nil { - InitTransactionManager("futriis.wal") - } - - tx := &Transaction{ - ID: TransactionID(globalTxManager.nextTxID.Add(1) - 1), - StartTime: time.Now().UnixMilli(), - Operations: make([]Operation, 0, 100), - Version: 1, - } - tx.State.Store(int32(TransactionActive)) - - globalTxManager.activeTransactions.Store(tx.ID, tx) - currentTx.Store(tx) - - AuditLog("START", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{ - "start_time": tx.StartTime, - }) - - return tx -} - -// BeginTransactionWithTimeout начинает транзакцию с таймаутом -func BeginTransactionWithTimeout(timeout time.Duration) *TransactionWithTimeout { - return NewTransactionWithTimeout(timeout) -} - -// BeginTransactionOnCollection начинает транзакцию для конкретной коллекции -func BeginTransactionOnCollection(coll *Collection) error { - if globalTxManager == nil { - if err := InitTransactionManager("futriis.wal"); err != nil { - return err - } - } - - tx := BeginTransaction() - if tx == nil { - return fmt.Errorf("failed to create transaction") - } - - if globalTxManager.logger != nil { - globalTxManager.logger.Debug(fmt.Sprintf("Transaction %d started on collection %s.%s", tx.ID, coll.dbName, coll.name)) - } - - return nil -} - -// CommitCurrentTransaction коммитит текущую транзакцию -func CommitCurrentTransaction() error { - txVal := currentTx.Load() - if txVal == nil { - return fmt.Errorf("no active transaction") - } - - tx := txVal.(*Transaction) - if TransactionState(tx.State.Load()) != TransactionActive { - return fmt.Errorf("transaction is not active") - } - - for _, op := range tx.Operations { - if err := applyOperation(op); err != nil { - AbortCurrentTransaction() - return fmt.Errorf("transaction commit failed at operation %s: %v", op.Type, err) - } - if globalTxManager != nil && op.DocumentID != "" { - if globalStorage != nil { - db, _ := globalStorage.GetDatabase(op.Database) - if db != nil { - coll, _ := db.GetCollection(op.Collection) - if coll != nil { - if doc, err := coll.Find(op.DocumentID); err == nil { - globalTxManager.AddDocumentVersion(op.DocumentID, &DocumentVersion{ - Document: doc.Clone(), - Timestamp: time.Now().UnixMilli(), - TxID: tx.ID, - }) - } - } - } - } - } - } - - tx.State.Store(int32(TransactionCommitted)) - - AuditLog("COMMIT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{ - "operations": len(tx.Operations), - }) - - currentTx.Store(nil) - globalTxManager.activeTransactions.Delete(tx.ID) - - return nil -} - -// AbortCurrentTransaction откатывает текущую транзакцию -func AbortCurrentTransaction() error { - txVal := currentTx.Load() - if txVal == nil { - return fmt.Errorf("no active transaction") - } - - tx := txVal.(*Transaction) - tx.State.Store(int32(TransactionAborted)) - - AuditLog("ABORT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{ - "operations": len(tx.Operations), - }) - - currentTx.Store(nil) - globalTxManager.activeTransactions.Delete(tx.ID) - - return nil -} - -// HasActiveTransaction проверяет наличие активной транзакции -func HasActiveTransaction() bool { - return currentTx.Load() != nil -} - -// GetCurrentTransactionID возвращает ID текущей транзакции -func GetCurrentTransactionID() string { - txVal := currentTx.Load() - if txVal == nil { - return "" - } - tx := txVal.(*Transaction) - return fmt.Sprintf("%d", tx.ID) -} - -// GetActiveTransactions возвращает список активных транзакций для API -func GetActiveTransactions() []TransactionInfo { - if globalTxManager == nil { - return []TransactionInfo{} - } - - transactions := make([]TransactionInfo, 0) - - globalTxManager.activeTransactions.Range(func(key, value interface{}) bool { - tx := value.(*Transaction) - status := "active" - if TransactionState(tx.State.Load()) == TransactionCommitted { - status = "committed" - } else if TransactionState(tx.State.Load()) == TransactionAborted { - status = "aborted" - } - - tx.mu.RLock() - opCount := len(tx.Operations) - operations := make([]OperationInfo, 0, opCount) - for _, op := range tx.Operations { - operations = append(operations, OperationInfo{ - Type: op.Type, - Database: op.Database, - Collection: op.Collection, - DocumentID: op.DocumentID, - }) - } - tx.mu.RUnlock() - - transactions = append(transactions, TransactionInfo{ - ID: fmt.Sprintf("%d", tx.ID), - Status: status, - StartTime: tx.StartTime, - OperationCount: opCount, - Operations: operations, - }) - return true - }) - - return transactions -} - -// GetTransactionByID возвращает транзакцию по ID -func GetTransactionByID(id string) (*Transaction, error) { - if globalTxManager == nil { - return nil, fmt.Errorf("transaction manager not initialized") - } - - var txID TransactionID - fmt.Sscanf(id, "%d", &txID) - - if val, ok := globalTxManager.activeTransactions.Load(txID); ok { - return val.(*Transaction), nil - } - - return nil, fmt.Errorf("transaction not found") -} - -// AddToTransaction добавляет операцию в текущую транзакцию -func AddToTransaction(coll *Collection, opType string, docID string, newData map[string]interface{}, oldData map[string]interface{}, version uint64) error { - txVal := currentTx.Load() - if txVal == nil { - return fmt.Errorf("no active transaction") - } - - tx := txVal.(*Transaction) - if TransactionState(tx.State.Load()) != TransactionActive { - return fmt.Errorf("transaction is not active") - } - - op := Operation{ - Type: opType, - Database: coll.dbName, - Collection: coll.name, - DocumentID: docID, - Data: newData, - OldData: oldData, - Version: version, - } - - tx.mu.Lock() - tx.Operations = append(tx.Operations, op) - tx.mu.Unlock() - - return nil -} - -// FindInTransaction ищет документ в контексте транзакции с поддержкой MVCC -func FindInTransaction(coll *Collection, id string) (*Document, error) { - txVal := currentTx.Load() - if txVal == nil { - return coll.Find(id) - } - - tx := txVal.(*Transaction) - - tx.mu.RLock() - defer tx.mu.RUnlock() - - for i := len(tx.Operations) - 1; i >= 0; i-- { - op := tx.Operations[i] - if op.DocumentID == id { - if op.Type == "delete" { - return nil, fmt.Errorf("document deleted in transaction") - } - if op.Type == "insert" || op.Type == "update" { - doc := NewDocumentWithID(op.DocumentID) - for k, v := range op.Data { - doc.SetField(k, v) - } - doc.Version = op.Version - return doc, nil - } - } - } - - if globalTxManager != nil { - if versionDoc := globalTxManager.GetDocumentVersion(id, tx.StartTime); versionDoc != nil { - return versionDoc, nil - } - } - - return coll.Find(id) -} - -// applyOperation применяет операцию к хранилищу -func applyOperation(op Operation) error { - if globalStorage == nil { - return fmt.Errorf("storage not initialized") - } - - db, err := globalStorage.GetDatabase(op.Database) - if err != nil { - return fmt.Errorf("database not found: %s", op.Database) - } - - coll, err := db.GetCollection(op.Collection) - if err != nil { - return fmt.Errorf("collection not found: %s", op.Collection) - } - - switch op.Type { - case "insert": - doc := NewDocumentWithID(op.DocumentID) - for k, v := range op.Data { - doc.SetField(k, v) - } - doc.Version = op.Version - return coll.Insert(doc) - - case "update": - return coll.Update(op.DocumentID, op.Data) - - case "delete": - return coll.Delete(op.DocumentID) - } - - return nil -} - -// SetGlobalStorage устанавливает глобальное хранилище -func SetGlobalStorage(s *Storage) { - globalStorage = s -} - -// GetGlobalStorage возвращает глобальное хранилище -func GetGlobalStorage() *Storage { - return globalStorage -} - -// AuditLog записывает событие в аудит -func AuditLog(operation, dataType, name string, details map[string]interface{}) { - LogAudit(operation, dataType, name, details) -} - -// AddDocumentVersion добавляет версию документа для MVCC -func (tm *TransactionManager) AddDocumentVersion(docID string, version *DocumentVersion) { - val, _ := tm.documentVersions.LoadOrStore(docID, make([]*DocumentVersion, 0)) - versions := val.([]*DocumentVersion) - versions = append(versions, version) - if len(versions) > tm.maxVersions && tm.maxVersions > 0 { - versions = versions[len(versions)-tm.maxVersions:] - } - tm.documentVersions.Store(docID, versions) - - if tm.visibilityMap != nil { - tm.visibilityMap.MarkVisible(docID, uint64(version.TxID), true) - } -} - -// GetDocumentVersion возвращает версию документа на определённый момент времени -func (tm *TransactionManager) GetDocumentVersion(docID string, timestamp int64) *Document { - if tm.readCache != nil { - if cached := tm.readCache.Get(docID, timestamp); cached != nil { - return cached - } - } - - val, ok := tm.documentVersions.Load(docID) - if !ok { - return nil - } - - versions := val.([]*DocumentVersion) - for i := len(versions) - 1; i >= 0; i-- { - if versions[i].Timestamp <= timestamp { - doc := versions[i].Document.Clone() - if tm.readCache != nil { - tm.readCache.Set(docID, timestamp, doc) - } - return doc - } - } - return nil -} - -// checkpointLoop периодически создаёт чекпоинты -func (tm *TransactionManager) checkpointLoop() { - ticker := time.NewTicker(time.Duration(tm.checkpointInterval) * time.Second) - defer ticker.Stop() - - for range ticker.C { - tm.createCheckpoint() - } -} - -// createCheckpoint создаёт чекпоинт текущего состояния -func (tm *TransactionManager) createCheckpoint() { - if tm.wal == nil { - return - } - - now := time.Now().Unix() - if now-tm.lastCheckpoint < tm.checkpointInterval { - return - } - - checkpointPath := fmt.Sprintf("%s.checkpoint.%d", tm.walPath, now) - - checkpoint := make(map[string]interface{}) - checkpoint["timestamp"] = now - checkpoint["last_lsn"] = tm.wal.currentSegment.EndLSN - - data, err := json.Marshal(checkpoint) - if err != nil { - if tm.logger != nil { - tm.logger.Error(fmt.Sprintf("Failed to marshal checkpoint: %v", err)) - } - return - } - - if err := os.WriteFile(checkpointPath, data, 0644); err != nil { - if tm.logger != nil { - tm.logger.Error(fmt.Sprintf("Failed to write checkpoint: %v", err)) - } - return - } - - tm.lastCheckpoint = now - if tm.logger != nil { - tm.logger.Info(fmt.Sprintf("Checkpoint created: %s", checkpointPath)) - } -} - -// versionCleanupLoop периодически очищает старые версии документов -func (tm *TransactionManager) versionCleanupLoop() { - if tm.maxVersions <= 0 { - return - } - - pruner := NewVersionPruner(&VersionPrunePolicy{ - MaxVersions: tm.maxVersions, - RetentionDays: VersionRetentionDays, - MinVersionAge: time.Hour, - PruneInterval: VersionPruneInterval, - }, tm, tm.logger) - pruner.Start() -} - -// MVCCSnapshot создаёт снапшот текущего состояния для MVCC -func MVCCSnapshot() uint64 { - return uint64(time.Now().UnixNano()) -} - -// CreateDocumentVersion создаёт новую версию документа для MVCC -func CreateDocumentVersion(doc *Document, txID TransactionID) *DocumentVersion { - return &DocumentVersion{ - Document: doc.Clone(), - Timestamp: time.Now().UnixMilli(), - TxID: txID, - } -} - -// StopTransactionManager останавливает менеджер транзакций -func StopTransactionManager() error { - if globalTxManager == nil { - return nil - } - - if globalTxManager.deadlockDetector != nil { - globalTxManager.deadlockDetector.Stop() - } - - if globalTxManager.wal != nil { - return globalTxManager.wal.Close() - } - - return nil -}