/* * 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 } // ========== Константы ========== 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 с повторными попытками (используем функции из fsync.go) 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 } // 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) } // 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(), } } // ========== 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(), } } // 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(), } } // ========== 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) *Transaction { tx := BeginTransaction() if tx == nil { return nil } time.AfterFunc(timeout, func() { if tx != nil && TransactionState(tx.State.Load()) == TransactionActive { tx.State.Store(int32(TransactionAborted)) if globalTxManager != nil { globalTxManager.activeTransactions.Delete(tx.ID) } AuditLog("TIMEOUT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{ "timeout_ms": timeout.Milliseconds(), }) } }) return tx } // 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, doc *Document) 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: doc.ID, Data: doc.GetFields(), Version: doc.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 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 } ticker := time.NewTicker(VersionPruneInterval) defer ticker.Stop() for range ticker.C { cutoffTime := time.Now().AddDate(0, 0, -VersionRetentionDays).UnixMilli() tm.documentVersions.Range(func(key, value interface{}) bool { versions := value.([]*DocumentVersion) if len(versions) <= tm.maxVersions { return true } newVersions := make([]*DocumentVersion, 0, tm.maxVersions) for _, v := range versions { if v.Timestamp >= cutoffTime && len(newVersions) < tm.maxVersions { newVersions = append(newVersions, v) } } if len(newVersions) < len(versions) { tm.documentVersions.Store(key, newVersions) } return true }) } } // 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 } // ========== 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)) } } } } // ========== Distributed Transaction Coordinator ========== // TxState представляет состояние транзакции type TxState int32 const ( TxActive TxState = iota TxPrepared TxCommitted TxAborted TxTimeout ) // DistributedTxInfo содержит информацию о распределённой транзакции type DistributedTxInfo struct { TxID TransactionID Nodes []string Prepared map[string]bool Committed map[string]bool Status TxState StartTime int64 Timeout time.Duration } // 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 } // ========== Дополнительные функции для совместимости ========== // 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, } }