/* * 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/transactions.go // Назначение: Реализация транзакций с поддержкой MVCC и WAL без блокировок. // Для распределённых транзакций используется протокол SAGA. // ИСПРАВЛЕНО: Deadlock в flushBatch/rotateSegment (wm.mu рекурсивно) // ИСПРАВЛЕНО: Race между async Write и Sync - добавлен механизм подтверждения // ИСПРАВЛЕНО: Apply до WAL - теперь сначала WAL, потом применение // ИСПРАВЛЕНО: Stop теряет записи буфера - flush перед закрытием package storage import ( "bufio" "encoding/binary" "encoding/json" "fmt" "io" "os" "path/filepath" "sort" "sync" "sync/atomic" "time" "futriis/internal/config" ) // ============================================================================= // БАЗОВЫЕ ТИПЫ // ============================================================================= type TransactionID uint64 type TransactionState int32 const ( TransactionActive TransactionState = iota TransactionCommitted TransactionAborted TransactionPrepared ) type TransactionRecord struct { ID TransactionID `json:"id"` State TransactionState `json:"state"` Timestamp int64 `json:"timestamp"` Operations []Operation `json:"operations"` IsDistributed bool `json:"is_distributed"` Nodes []string `json:"nodes,omitempty"` } type WALRecord struct { CRC uint32 `json:"crc"` Length uint32 `json:"length"` Type byte `json:"type"` Data []byte `json:"data"` Timestamp int64 `json:"timestamp"` LSN uint64 `json:"lsn"` } // ============================================================================= // КОНСТАНТЫ // ============================================================================= const ( WALSegmentSize = 64 * 1024 * 1024 WALSegmentPrefix = "wal_segment_" WALIndexPrefix = "wal_index_" VersionPruneInterval = 5 * time.Minute DefaultTxTimeout = 30 * time.Second DeadlockCheckInterval = 1 * time.Second MaxSavepointsPerTx = 100 AsyncRecoveryBufferSize = 10000 AsyncRecoveryWorkers = 4 AsyncRecoveryTimeout = 30 * time.Second FsyncMaxRetries = 3 FsyncRetryDelay = 100 * time.Millisecond SagaStateDir = "saga_states" SagaStateFilePrefix = "saga_state_" SagaStateFileSuffix = ".json" SagaCheckpointInterval = 30 * time.Second SagaMaxRetries = 5 SagaRetryBackoff = 100 * time.Millisecond SagaOrphanTimeout = 5 * time.Minute SagaCleanupInterval = 1 * time.Minute ) // ============================================================================= // 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, 0x37d83bf8, 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9, 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6, 0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf, 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94, 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d, } func crc32(data []byte) uint32 { crc := uint32(0xFFFFFFFF) for _, b := range data { crc = (crc >> 8) ^ crc32Table[(crc^uint32(b))&0xFF] } return crc ^ 0xFFFFFFFF } // ============================================================================= // АУДИТ ТРАНЗАКЦИЙ // ============================================================================= type TransactionAuditEntry struct { TxID TransactionID `json:"tx_id"` Action string `json:"action"` State TransactionState `json:"state"` Timestamp int64 `json:"timestamp"` TimestampStr string `json:"timestamp_str"` Details map[string]interface{} `json:"details"` } type TransactionAuditLogger struct { entries []TransactionAuditEntry mu sync.RWMutex maxSize int filePath string fileMu sync.Mutex enabled bool } var globalTxAuditLogger = &TransactionAuditLogger{ entries: make([]TransactionAuditEntry, 0), maxSize: 100000, enabled: true, } func InitTransactionAuditLogger(filePath string) error { globalTxAuditLogger.filePath = filePath if filePath != "" { if err := os.MkdirAll(filepath.Dir(filePath), 0755); err != nil { return err } globalTxAuditLogger.loadFromFile() } return nil } func (tal *TransactionAuditLogger) loadFromFile() { if tal.filePath == "" { return } data, err := os.ReadFile(tal.filePath) if err != nil { return } var entries []TransactionAuditEntry if err := json.Unmarshal(data, &entries); err != nil { return } tal.mu.Lock() defer tal.mu.Unlock() tal.entries = entries } func (tal *TransactionAuditLogger) saveToFile() { if tal.filePath == "" { return } tal.fileMu.Lock() defer tal.fileMu.Unlock() tal.mu.RLock() data, err := json.MarshalIndent(tal.entries, "", " ") tal.mu.RUnlock() if err != nil { return } os.WriteFile(tal.filePath, data, 0644) } func LogTransactionAudit(txID TransactionID, action string, state TransactionState, details map[string]interface{}) { if !globalTxAuditLogger.enabled { return } now := time.Now() entry := TransactionAuditEntry{ TxID: txID, Action: action, State: state, Timestamp: now.UnixMilli(), TimestampStr: now.Format("2006-01-02 15:04:05.000"), Details: details, } globalTxAuditLogger.mu.Lock() globalTxAuditLogger.entries = append(globalTxAuditLogger.entries, entry) if len(globalTxAuditLogger.entries) > globalTxAuditLogger.maxSize { globalTxAuditLogger.entries = globalTxAuditLogger.entries[len(globalTxAuditLogger.entries)-globalTxAuditLogger.maxSize:] } globalTxAuditLogger.mu.Unlock() go globalTxAuditLogger.saveToFile() } func GetTransactionAuditLog() []TransactionAuditEntry { globalTxAuditLogger.mu.RLock() defer globalTxAuditLogger.mu.RUnlock() result := make([]TransactionAuditEntry, len(globalTxAuditLogger.entries)) copy(result, globalTxAuditLogger.entries) return result } // ============================================================================= // MVCC MANAGER // ============================================================================= type MVCCManager struct { versions sync.Map visibilityMap *VisibilityMap readCache *ReadTimestampCache maxVersionsPerDoc int retentionDuration time.Duration stats MVCCStats mu sync.RWMutex activeReaders sync.Map readersMu sync.RWMutex oldestReadTx atomic.Int64 } type MVCCStats struct { TotalVersionsCreated atomic.Uint64 TotalVersionsPruned atomic.Uint64 TotalCacheHits atomic.Uint64 TotalCacheMisses atomic.Uint64 TotalVisibilityHits atomic.Uint64 TotalVisibilityMisses atomic.Uint64 TotalSkippedPrune atomic.Uint64 } func NewMVCCManager(maxVersionsPerDoc int, retentionDays int) *MVCCManager { if maxVersionsPerDoc <= 0 { maxVersionsPerDoc = MaxVersionsPerDoc } if retentionDays <= 0 { retentionDays = VersionRetentionDays } m := &MVCCManager{ maxVersionsPerDoc: maxVersionsPerDoc, retentionDuration: time.Duration(retentionDays) * 24 * time.Hour, visibilityMap: NewVisibilityMap(VisibilityMapSize), readCache: NewReadTimestampCache(10000, 5*time.Minute), } m.oldestReadTx.Store(0) go m.pruneOldVersionsLoop() return m } func (m *MVCCManager) RegisterReader(txID uint64) { m.readersMu.Lock() defer m.readersMu.Unlock() m.activeReaders.Store(txID, time.Now().UnixMilli()) m.updateOldestReader() } func (m *MVCCManager) UnregisterReader(txID uint64) { m.readersMu.Lock() defer m.readersMu.Unlock() m.activeReaders.Delete(txID) m.updateOldestReader() } func (m *MVCCManager) updateOldestReader() { oldest := int64(^uint64(0) >> 1) m.activeReaders.Range(func(key, value interface{}) bool { if ts, ok := value.(int64); ok && ts < oldest { oldest = ts } return true }) m.oldestReadTx.Store(oldest) } func (m *MVCCManager) CreateVersion(doc *Document, txID TransactionID) *DocumentVersion { version := &DocumentVersion{ Document: doc.Clone(), Timestamp: time.Now().UnixMilli(), TxID: txID, VersionID: fmt.Sprintf("%s_%d_%d", doc.ID, txID, time.Now().UnixNano()), } m.mu.Lock() defer m.mu.Unlock() val, _ := m.versions.LoadOrStore(doc.ID, make([]*DocumentVersion, 0)) versions := val.([]*DocumentVersion) versions = append(versions, version) if len(versions) > m.maxVersionsPerDoc { oldestReader := m.oldestReadTx.Load() newVersions := make([]*DocumentVersion, 0, m.maxVersionsPerDoc) startIdx := len(versions) - m.maxVersionsPerDoc if startIdx < 0 { startIdx = 0 } for i := startIdx; i < len(versions); i++ { newVersions = append(newVersions, versions[i]) } if oldestReader > 0 { for i := 0; i < startIdx; i++ { if versions[i].Timestamp >= oldestReader { newVersions = append([]*DocumentVersion{versions[i]}, newVersions...) } } } prunedCount := len(versions) - len(newVersions) versions = newVersions m.stats.TotalVersionsPruned.Add(uint64(prunedCount)) } m.versions.Store(doc.ID, versions) m.visibilityMap.MarkVisible(doc.ID, uint64(txID), true) m.stats.TotalVersionsCreated.Add(1) LogTransactionAudit(txID, "VERSION_CREATE", TransactionActive, map[string]interface{}{ "doc_id": doc.ID, "version": version.VersionID, }) return version } func (m *MVCCManager) GetVersionAt(docID string, timestamp int64) *Document { if cached := m.readCache.Get(docID, timestamp); cached != nil { m.stats.TotalCacheHits.Add(1) return cached } m.stats.TotalCacheMisses.Add(1) m.mu.RLock() defer m.mu.RUnlock() val, ok := m.versions.Load(docID) if !ok { return nil } versions := val.([]*DocumentVersion) var result *DocumentVersion for i := len(versions) - 1; i >= 0; i-- { v := versions[i] if v.Timestamp <= timestamp && m.visibilityMap.IsVisible(docID, uint64(v.TxID)) { result = v break } } if result == nil { return nil } doc := result.Document.Clone() m.readCache.Set(docID, timestamp, doc) return doc } func (m *MVCCManager) GetLatestVersion(docID string) *Document { m.mu.RLock() defer m.mu.RUnlock() val, ok := m.versions.Load(docID) if !ok { return nil } versions := val.([]*DocumentVersion) if len(versions) == 0 { return nil } return versions[len(versions)-1].Document.Clone() } func (m *MVCCManager) GetAllVersions(docID string) []*DocumentVersion { m.mu.RLock() defer m.mu.RUnlock() val, ok := m.versions.Load(docID) if !ok { return nil } versions := val.([]*DocumentVersion) result := make([]*DocumentVersion, len(versions)) for i, v := range versions { result[i] = &DocumentVersion{ Document: v.Document.Clone(), Timestamp: v.Timestamp, TxID: v.TxID, VersionID: v.VersionID, } } return result } func (m *MVCCManager) pruneOldVersionsLoop() { ticker := time.NewTicker(VersionPruneInterval) defer ticker.Stop() for range ticker.C { m.PruneOldVersions() } } func (m *MVCCManager) PruneOldVersions() { cutoffTime := time.Now().Add(-m.retentionDuration).UnixMilli() oldestReader := m.oldestReadTx.Load() pruned := int64(0) m.mu.Lock() defer m.mu.Unlock() m.versions.Range(func(key, value interface{}) bool { docID := key.(string) versions := value.([]*DocumentVersion) newVersions := make([]*DocumentVersion, 0, len(versions)) for _, v := range versions { if v.Timestamp >= cutoffTime || (oldestReader > 0 && v.Timestamp >= oldestReader) { newVersions = append(newVersions, v) } else { pruned++ } } if len(newVersions) == 0 && len(versions) > 0 { newVersions = append(newVersions, versions[len(versions)-1]) pruned-- } if len(newVersions) != len(versions) { m.versions.Store(docID, newVersions) } return true }) if pruned > 0 { m.stats.TotalVersionsPruned.Add(uint64(pruned)) } } func (m *MVCCManager) GetMVCCStats() map[string]interface{} { return map[string]interface{}{ "total_versions_created": m.stats.TotalVersionsCreated.Load(), "total_versions_pruned": m.stats.TotalVersionsPruned.Load(), "total_cache_hits": m.stats.TotalCacheHits.Load(), "total_cache_misses": m.stats.TotalCacheMisses.Load(), "total_visibility_hits": m.stats.TotalVisibilityHits.Load(), "total_visibility_misses": m.stats.TotalVisibilityMisses.Load(), "total_skipped_prune": m.stats.TotalSkippedPrune.Load(), "max_versions_per_doc": m.maxVersionsPerDoc, "retention_days": int(m.retentionDuration.Hours() / 24), "oldest_read_tx": m.oldestReadTx.Load(), } } // ============================================================================= // VISIBILITY MAP // ============================================================================= type VisibilityMapEntry struct { DocID string VisibleFrom uint64 VisibleTo uint64 IsVisible bool LastAccess int64 } type VisibilityMap struct { entries sync.Map maxSize int hitCount atomic.Uint64 missCount atomic.Uint64 mu sync.RWMutex } func NewVisibilityMap(maxSize int) *VisibilityMap { if maxSize <= 0 { maxSize = VisibilityMapSize } return &VisibilityMap{maxSize: maxSize} } 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(), }) } func (vm *VisibilityMap) IsVisible(docID string, version uint64) bool { key := fmt.Sprintf("%s@%d", docID, version) if val, ok := vm.entries.Load(key); ok { entry := val.(*VisibilityMapEntry) entry.LastAccess = time.Now().Unix() vm.hitCount.Add(1) return entry.IsVisible } var found bool vm.entries.Range(func(k, v interface{}) bool { entry := v.(*VisibilityMapEntry) if entry.DocID == docID && version >= entry.VisibleFrom && version <= entry.VisibleTo && entry.IsVisible { found = true return false } return true }) if found { vm.hitCount.Add(1) return true } vm.missCount.Add(1) return false } func (vm *VisibilityMap) GetStats() map[string]interface{} { count := 0 vm.entries.Range(func(_, _ interface{}) bool { count++ return true }) return map[string]interface{}{ "hits": vm.hitCount.Load(), "misses": vm.missCount.Load(), "entries": count, "max_size": vm.maxSize, } } // ============================================================================= // READ TIMESTAMP CACHE // ============================================================================= type ReadTimestampCache struct { cache sync.Map maxSize int ttl time.Duration hits atomic.Uint64 misses atomic.Uint64 size atomic.Int64 mu sync.RWMutex } type cachedEntry struct { doc *Document cachedAt time.Time accessCount int64 } func NewReadTimestampCache(maxSize int, ttl time.Duration) *ReadTimestampCache { if maxSize <= 0 { maxSize = 10000 } if ttl <= 0 { ttl = 5 * time.Minute } c := &ReadTimestampCache{maxSize: maxSize, ttl: ttl} go c.cleanupLoop() return c } 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.size.Add(-1) rtc.misses.Add(1) return nil } entry.accessCount++ rtc.hits.Add(1) return entry.doc } func (rtc *ReadTimestampCache) Set(docID string, timestamp int64, doc *Document) { key := fmt.Sprintf("%s@%d", docID, timestamp) if rtc.size.Load() >= int64(rtc.maxSize) { rtc.evictOldest() } rtc.cache.Store(key, &cachedEntry{doc: doc, cachedAt: time.Now(), accessCount: 0}) rtc.size.Add(1) } func (rtc *ReadTimestampCache) evictOldest() { var oldestKey interface{} var oldestTime time.Time rtc.cache.Range(func(key, value interface{}) bool { entry := value.(*cachedEntry) if oldestKey == nil || entry.cachedAt.Before(oldestTime) { oldestKey = key oldestTime = entry.cachedAt } return true }) if oldestKey != nil { rtc.cache.Delete(oldestKey) rtc.size.Add(-1) } } func (rtc *ReadTimestampCache) cleanupLoop() { ticker := time.NewTicker(rtc.ttl) defer ticker.Stop() for range ticker.C { rtc.cache.Range(func(key, value interface{}) bool { entry := value.(*cachedEntry) if time.Since(entry.cachedAt) > rtc.ttl { rtc.cache.Delete(key) rtc.size.Add(-1) } return true }) } } func (rtc *ReadTimestampCache) GetStats() map[string]interface{} { return map[string]interface{}{ "hits": rtc.hits.Load(), "misses": rtc.misses.Load(), "size": rtc.size.Load(), "max_size": rtc.maxSize, "ttl_secs": int(rtc.ttl.Seconds()), } } // ============================================================================= // SAGA PERSISTENT STORAGE // ============================================================================= type SagaState struct { ID string `json:"id"` Status string `json:"status"` CurrentStep int `json:"current_step"` Steps []SagaStepState `json:"steps"` Data map[string]interface{} `json:"data"` CreatedAt int64 `json:"created_at"` UpdatedAt int64 `json:"updated_at"` CompletedAt int64 `json:"completed_at,omitempty"` CompensationExecuted bool `json:"compensation_executed"` NodeID string `json:"node_id"` CoordinatorID string `json:"coordinator_id"` Version uint64 `json:"version"` RetryCount int `json:"retry_count"` LastError string `json:"last_error,omitempty"` ExecutionID string `json:"execution_id"` } type SagaStepState struct { ID string `json:"id"` Name string `json:"name"` Status string `json:"status"` Data map[string]interface{} `json:"data"` StartedAt int64 `json:"started_at"` CompletedAt int64 `json:"completed_at"` ExecutionID string `json:"execution_id"` RetryCount int `json:"retry_count"` LastError string `json:"last_error,omitempty"` CompensatedAt int64 `json:"compensated_at,omitempty"` } type SagaPersistentStorage struct { baseDir string mu sync.RWMutex cache map[string]*SagaState maxCache int logger LoggerInterface fsyncEnabled bool fsyncMaxRetries int fsyncRetryDelay time.Duration } func NewSagaPersistentStorage(baseDir string, logger LoggerInterface) (*SagaPersistentStorage, error) { return NewSagaPersistentStorageWithConfig(nil, logger) } func NewSagaPersistentStorageWithConfig(cfg *config.SagaConfig, logger LoggerInterface) (*SagaPersistentStorage, error) { baseDir := SagaStateDir maxCache := 10000 fsyncEnabled := true fsyncMaxRetries := 3 fsyncRetryDelay := 100 * time.Millisecond if cfg != nil { baseDir = cfg.GetStateDir() maxCache = cfg.GetMaxCacheSize() fsyncEnabled = cfg.IsFsyncEnabled() fsyncMaxRetries = cfg.GetFsyncMaxRetries() fsyncRetryDelay = cfg.GetFsyncRetryDelay() } fullPath := filepath.Join(baseDir) if err := os.MkdirAll(fullPath, 0755); err != nil { return nil, fmt.Errorf("failed to create saga state directory: %v", err) } return &SagaPersistentStorage{ baseDir: fullPath, cache: make(map[string]*SagaState), maxCache: maxCache, logger: logger, fsyncEnabled: fsyncEnabled, fsyncMaxRetries: fsyncMaxRetries, fsyncRetryDelay: fsyncRetryDelay, }, nil } func (sps *SagaPersistentStorage) getStatePath(sagaID string) string { return filepath.Join(sps.baseDir, fmt.Sprintf("%s%s%s", SagaStateFilePrefix, sagaID, SagaStateFileSuffix)) } func (sps *SagaPersistentStorage) Save(state *SagaState) error { sps.mu.Lock() defer sps.mu.Unlock() state.UpdatedAt = time.Now().UnixMilli() state.Version++ if len(sps.cache) >= sps.maxCache { var oldestKey string var oldestTime int64 = time.Now().UnixMilli() for k, v := range sps.cache { if v.UpdatedAt < oldestTime { oldestTime = v.UpdatedAt oldestKey = k } } if oldestKey != "" { delete(sps.cache, oldestKey) } } sps.cache[state.ID] = state path := sps.getStatePath(state.ID) data, err := json.MarshalIndent(state, "", " ") if err != nil { return fmt.Errorf("failed to marshal saga state: %v", err) } tmpPath := path + ".tmp" if err := os.WriteFile(tmpPath, data, 0644); err != nil { return fmt.Errorf("failed to write saga state: %v", err) } if sps.fsyncEnabled { if f, err := os.OpenFile(tmpPath, os.O_RDWR, 0644); err == nil { for i := 0; i < sps.fsyncMaxRetries; i++ { if err := RealFsyncWithRetry(f, sps.fsyncMaxRetries, sps.fsyncRetryDelay); err == nil { break } if i < sps.fsyncMaxRetries-1 { time.Sleep(sps.fsyncRetryDelay) } } f.Close() } } if err := os.Rename(tmpPath, path); err != nil { return fmt.Errorf("failed to rename saga state: %v", err) } if sps.fsyncEnabled { FsyncDir(sps.baseDir) } if sps.logger != nil { sps.logger.Debug(fmt.Sprintf("Saved saga state %s (version %d, status %s)", state.ID, state.Version, state.Status)) } return nil } func (sps *SagaPersistentStorage) Load(sagaID string) (*SagaState, error) { sps.mu.RLock() if state, ok := sps.cache[sagaID]; ok { sps.mu.RUnlock() return state, nil } sps.mu.RUnlock() path := sps.getStatePath(sagaID) data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { return nil, nil } return nil, fmt.Errorf("failed to read saga state: %v", err) } var state SagaState if err := json.Unmarshal(data, &state); err != nil { return nil, fmt.Errorf("failed to unmarshal saga state: %v", err) } sps.mu.Lock() sps.cache[sagaID] = &state sps.mu.Unlock() return &state, nil } func (sps *SagaPersistentStorage) Delete(sagaID string) error { sps.mu.Lock() delete(sps.cache, sagaID) sps.mu.Unlock() path := sps.getStatePath(sagaID) if err := os.Remove(path); err != nil && !os.IsNotExist(err) { return fmt.Errorf("failed to delete saga state: %v", err) } return nil } func (sps *SagaPersistentStorage) ListAll() ([]*SagaState, error) { pattern := filepath.Join(sps.baseDir, fmt.Sprintf("%s*%s", SagaStateFilePrefix, SagaStateFileSuffix)) files, err := filepath.Glob(pattern) if err != nil { return nil, err } states := make([]*SagaState, 0, len(files)) for _, file := range files { data, err := os.ReadFile(file) if err != nil { continue } var state SagaState if err := json.Unmarshal(data, &state); err != nil { continue } states = append(states, &state) } return states, nil } func (sps *SagaPersistentStorage) ListPending() ([]*SagaState, error) { all, err := sps.ListAll() if err != nil { return nil, err } pending := make([]*SagaState, 0) for _, state := range all { if state.Status == "pending" || state.Status == "running" || state.Status == "compensating" { pending = append(pending, state) } } return pending, nil } // ============================================================================= // SAGA ORCHESTRATOR // ============================================================================= type SagaOrchestrator struct { storage *SagaPersistentStorage coordinators []*SagaCoordinator mu sync.RWMutex logger LoggerInterface nodeID string stopChan chan struct{} wg sync.WaitGroup isLeader atomic.Bool leaderID string electionMu sync.Mutex config *config.SagaConfig metrics *SagaMetrics } type SagaCoordinator struct { id string orchestrator *SagaOrchestrator activeSagas sync.Map stopChan chan struct{} wg sync.WaitGroup isActive bool config *config.SagaConfig mu sync.RWMutex } type SagaMetrics struct { TotalStarted atomic.Uint64 TotalCompleted atomic.Uint64 TotalAborted atomic.Uint64 TotalFailed atomic.Uint64 TotalCompensated atomic.Uint64 ActiveCount atomic.Int64 AvgDurationMs atomic.Uint64 TotalDurationMs atomic.Uint64 RecoveryCount atomic.Uint64 mu sync.RWMutex latencies []int64 maxLatency int64 minLatency int64 } func NewSagaOrchestrator(baseDir string, nodeID string, logger LoggerInterface) (*SagaOrchestrator, error) { return NewSagaOrchestratorWithConfig(nil, nodeID, logger) } func NewSagaOrchestratorWithConfig(cfg *config.SagaConfig, nodeID string, logger LoggerInterface) (*SagaOrchestrator, error) { if cfg == nil { cfg = &config.SagaConfig{ Enabled: true, CoordinatorCount: 3, StateDir: "saga_states", MaxRetries: 5, RetryBackoffMs: 100, SagaTimeoutSec: 300, StuckCheckIntervalSec: 10, LeaderElectionIntervalSec: 5, RecoveryIntervalSec: 30, MetricsIntervalSec: 60, CleanupPeriodHours: 24, MaxCacheSize: 10000, OperationRetentionDays: 7, ChannelBufferSize: 10000, AsyncRecoveryWorkers: 4, AsyncRecoveryTimeoutSec: 30, FsyncEnabled: true, FsyncMaxRetries: 3, FsyncRetryDelayMs: 100, } } storage, err := NewSagaPersistentStorageWithConfig(cfg, logger) if err != nil { return nil, err } o := &SagaOrchestrator{ storage: storage, logger: logger, nodeID: nodeID, stopChan: make(chan struct{}, cfg.GetChannelBufferSize()), metrics: &SagaMetrics{minLatency: -1}, coordinators: make([]*SagaCoordinator, 0), config: cfg, } coordinatorCount := cfg.GetCoordinatorCount() for i := 0; i < coordinatorCount; i++ { coord := &SagaCoordinator{ id: fmt.Sprintf("%s-coord-%d", nodeID, i), orchestrator: o, stopChan: make(chan struct{}, cfg.GetChannelBufferSize()), isActive: true, config: cfg, } o.coordinators = append(o.coordinators, coord) coord.wg.Add(1) go coord.run() } o.wg.Add(1) go o.leaderElectionLoop() o.wg.Add(1) go o.recoveryLoop() o.wg.Add(1) go o.metricsLoop() o.wg.Add(1) go o.orphanCleanupLoop() if cfg.IsSagaEnabled() && logger != nil { logger.Info(fmt.Sprintf("Saga orchestrator initialized on node %s with %d coordinators", nodeID, coordinatorCount)) } return o, nil } func (o *SagaOrchestrator) leaderElectionLoop() { defer o.wg.Done() ticker := time.NewTicker(o.config.GetLeaderElectionInterval()) defer ticker.Stop() for { select { case <-ticker.C: o.electLeader() case <-o.stopChan: return } } } func (o *SagaOrchestrator) electLeader() { o.electionMu.Lock() defer o.electionMu.Unlock() candidates := make([]string, 0) o.mu.RLock() for _, coord := range o.coordinators { if coord.isActive { candidates = append(candidates, coord.id) } } o.mu.RUnlock() if len(candidates) == 0 { o.isLeader.Store(false) o.leaderID = "" return } sort.Strings(candidates) leader := candidates[0] o.leaderID = leader isLeader := leader == o.coordinators[0].id o.isLeader.Store(isLeader) } func (o *SagaOrchestrator) IsLeader() bool { return o.isLeader.Load() } func (o *SagaOrchestrator) GetLeaderID() string { return o.leaderID } func (o *SagaOrchestrator) recoveryLoop() { defer o.wg.Done() ticker := time.NewTicker(o.config.GetRecoveryInterval()) defer ticker.Stop() for { select { case <-ticker.C: if o.IsLeader() { o.recoverPendingSagas() } case <-o.stopChan: return } } } func (o *SagaOrchestrator) orphanCleanupLoop() { defer o.wg.Done() ticker := time.NewTicker(SagaCleanupInterval) defer ticker.Stop() for { select { case <-ticker.C: if o.IsLeader() { o.cleanupOrphanSteps() } case <-o.stopChan: return } } } func (o *SagaOrchestrator) cleanupOrphanSteps() { pending, err := o.storage.ListPending() if err != nil { if o.logger != nil { o.logger.Error(fmt.Sprintf("Failed to list pending sagas for cleanup: %v", err)) } return } now := time.Now().UnixMilli() for _, state := range pending { if state.Status == "running" || state.Status == "compensating" { elapsed := now - state.UpdatedAt if elapsed > SagaOrphanTimeout.Milliseconds() { if o.logger != nil { o.logger.Warn(fmt.Sprintf("Found orphan saga %s (last updated %d ms ago), cleaning up", state.ID, elapsed)) } state.Status = "orphaned" state.LastError = fmt.Sprintf("orphaned after %d ms of inactivity", elapsed) o.storage.Save(state) o.compensateOrphanSaga(state) } } } } func (o *SagaOrchestrator) compensateOrphanSaga(state *SagaState) { if state.CompensationExecuted { return } if o.logger != nil { o.logger.Info(fmt.Sprintf("Compensating orphan saga %s", state.ID)) } saga := &SagaTransaction{ ID: state.ID, Status: state.Status, CurrentStep: state.CurrentStep, Data: state.Data, CreatedAt: state.CreatedAt, UpdatedAt: state.UpdatedAt, executedOps: make(map[string]bool), } for _, stepState := range state.Steps { step := &SagaStep{ ID: stepState.ID, Name: stepState.Name, Status: stepState.Status, Data: stepState.Data, StartedAt: stepState.StartedAt, CompletedAt: stepState.CompletedAt, ExecutionID: stepState.ExecutionID, RetryCount: stepState.RetryCount, LastError: stepState.LastError, } saga.Steps = append(saga.Steps, step) if stepState.Status == "completed" { saga.executedOps[stepState.ExecutionID] = true } } if err := o.executeCompensation(saga, len(saga.Steps)-1); err != nil { if o.logger != nil { o.logger.Error(fmt.Sprintf("Failed to compensate orphan saga %s: %v", state.ID, err)) } } state.CompensationExecuted = true state.Status = "orphaned_compensated" o.storage.Save(state) } func (o *SagaOrchestrator) recoverPendingSagas() { pending, err := o.storage.ListPending() if err != nil { if o.logger != nil { o.logger.Error(fmt.Sprintf("Failed to list pending sagas: %v", err)) } return } if len(pending) == 0 { return } if o.logger != nil { o.logger.Info(fmt.Sprintf("Recovering %d pending sagas", len(pending))) } for _, state := range pending { if state.NodeID != "" && state.NodeID != o.nodeID { if !o.isNodeAlive(state.NodeID) { if o.logger != nil { o.logger.Warn(fmt.Sprintf("Node %s is dead, taking over saga %s", state.NodeID, state.ID)) } state.NodeID = o.nodeID o.storage.Save(state) } else { continue } } if err := o.resumeSaga(state); err != nil { if o.logger != nil { o.logger.Error(fmt.Sprintf("Failed to resume saga %s: %v", state.ID, err)) } } else { o.metrics.RecoveryCount.Add(1) } } } func (o *SagaOrchestrator) isNodeAlive(nodeID string) bool { return nodeID == o.nodeID } func (o *SagaOrchestrator) resumeSaga(state *SagaState) error { saga := &SagaTransaction{ ID: state.ID, Status: state.Status, CurrentStep: state.CurrentStep, Data: state.Data, CreatedAt: state.CreatedAt, UpdatedAt: state.UpdatedAt, compensationExecuted: state.CompensationExecuted, executedOps: make(map[string]bool), } saga.Steps = make([]*SagaStep, len(state.Steps)) for i, stepState := range state.Steps { saga.Steps[i] = &SagaStep{ ID: stepState.ID, Name: stepState.Name, Status: stepState.Status, Data: stepState.Data, StartedAt: stepState.StartedAt, CompletedAt: stepState.CompletedAt, ExecutionID: stepState.ExecutionID, RetryCount: stepState.RetryCount, LastError: stepState.LastError, } if stepState.Status == "completed" { saga.executedOps[stepState.ExecutionID] = true } } o.mu.Lock() for _, coord := range o.coordinators { coord.activeSagas.Store(state.ID, saga) } o.mu.Unlock() return o.executeSagaInternal(saga) } func (o *SagaOrchestrator) metricsLoop() { defer o.wg.Done() ticker := time.NewTicker(o.config.GetMetricsInterval()) defer ticker.Stop() for { select { case <-ticker.C: if o.logger != nil { metrics := o.GetMetrics() o.logger.Debug(fmt.Sprintf("Saga metrics: started=%d, completed=%d, aborted=%d, active=%d", metrics["total_started"], metrics["total_completed"], metrics["total_aborted"], metrics["active_count"])) } case <-o.stopChan: return } } } func (o *SagaOrchestrator) BeginSaga(id string) (*SagaTransaction, error) { if !o.IsLeader() { return nil, fmt.Errorf("current node is not the leader") } existing, err := o.storage.Load(id) if err != nil { return nil, err } if existing != nil { return nil, fmt.Errorf("saga %s already exists", id) } now := time.Now().UnixMilli() state := &SagaState{ ID: id, Status: "pending", CurrentStep: 0, Data: make(map[string]interface{}), CreatedAt: now, UpdatedAt: now, NodeID: o.nodeID, CoordinatorID: o.coordinators[0].id, Version: 1, ExecutionID: fmt.Sprintf("%s_%d", id, now), } if err := o.storage.Save(state); err != nil { return nil, err } saga := &SagaTransaction{ ID: id, Steps: make([]*SagaStep, 0), CurrentStep: 0, Status: "pending", CreatedAt: now, UpdatedAt: now, Data: make(map[string]interface{}), executedOps: make(map[string]bool), } o.mu.RLock() for _, coord := range o.coordinators { coord.activeSagas.Store(id, saga) } o.mu.RUnlock() o.metrics.TotalStarted.Add(1) o.metrics.ActiveCount.Add(1) LogTransactionAudit(TransactionID(0), "SAGA_BEGIN", TransactionActive, map[string]interface{}{ "saga_id": id, "node_id": o.nodeID, }) if o.logger != nil { o.logger.Info(fmt.Sprintf("Saga %s started on node %s", id, o.nodeID)) } return saga, nil } func (o *SagaOrchestrator) Execute(saga *SagaTransaction) error { if !o.IsLeader() { return fmt.Errorf("current node is not the leader") } return o.executeSagaInternal(saga) } func (o *SagaOrchestrator) executeSagaInternal(saga *SagaTransaction) error { saga.mu.Lock() defer saga.mu.Unlock() if saga.Status != "pending" && saga.Status != "running" { return fmt.Errorf("saga %s is not in pending or running state", saga.ID) } saga.Status = "running" saga.UpdatedAt = time.Now().UnixMilli() if err := o.saveSagaState(saga); err != nil { return err } startTime := time.Now() for i := saga.CurrentStep; i < len(saga.Steps); i++ { step := saga.Steps[i] saga.CurrentStep = i if step.Status == "completed" { continue } step.Status = "running" step.StartedAt = time.Now().UnixMilli() if saga.HasExecuted(step.ExecutionID) { step.Status = "completed" step.CompletedAt = time.Now().UnixMilli() continue } var err error maxRetries := SagaMaxRetries if o.config != nil { maxRetries = o.config.GetMaxRetries() } for retry := 0; retry < maxRetries; retry++ { step.RetryCount = retry + 1 if err = step.Execute(); err == nil { break } step.LastError = err.Error() backoff := time.Duration(100*(1<= 0; j-- { step := saga.Steps[j] if step.Status == "compensated" || step.Status == "pending" { continue } if step.Status == "completed" { var err error maxRetries := SagaMaxRetries if o.config != nil { maxRetries = o.config.GetMaxRetries() } for retry := 0; retry < maxRetries; retry++ { if err = step.Compensate(); err == nil { break } backoff := time.Duration(100*(1< o.metrics.maxLatency { o.metrics.maxLatency = durationMs } o.metrics.latencies = append(o.metrics.latencies, durationMs) if len(o.metrics.latencies) > 1000 { o.metrics.latencies = o.metrics.latencies[1:] } } func (o *SagaOrchestrator) GetMetrics() map[string]interface{} { o.metrics.mu.RLock() defer o.metrics.mu.RUnlock() latencies := make([]int64, len(o.metrics.latencies)) copy(latencies, o.metrics.latencies) sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) p50, p95, p99 := int64(0), int64(0), int64(0) if len(latencies) > 0 { p50 = latencies[int(float64(len(latencies))*0.5)] p95 = latencies[int(float64(len(latencies))*0.95)] p99 = latencies[int(float64(len(latencies))*0.99)] } return map[string]interface{}{ "total_started": o.metrics.TotalStarted.Load(), "total_completed": o.metrics.TotalCompleted.Load(), "total_aborted": o.metrics.TotalAborted.Load(), "total_failed": o.metrics.TotalFailed.Load(), "total_compensated": o.metrics.TotalCompensated.Load(), "active_count": o.metrics.ActiveCount.Load(), "avg_duration_ms": o.metrics.AvgDurationMs.Load(), "min_duration_ms": o.metrics.minLatency, "max_duration_ms": o.metrics.maxLatency, "p50_duration_ms": p50, "p95_duration_ms": p95, "p99_duration_ms": p99, "recovery_count": o.metrics.RecoveryCount.Load(), "is_leader": o.IsLeader(), "leader_id": o.GetLeaderID(), "node_id": o.nodeID, } } func (o *SagaOrchestrator) GetSaga(id string) (*SagaTransaction, error) { o.mu.RLock() for _, coord := range o.coordinators { if val, ok := coord.activeSagas.Load(id); ok { o.mu.RUnlock() return val.(*SagaTransaction), nil } } o.mu.RUnlock() state, err := o.storage.Load(id) if err != nil { return nil, err } if state == nil { return nil, fmt.Errorf("saga %s not found", id) } saga := &SagaTransaction{ ID: state.ID, Status: state.Status, CurrentStep: state.CurrentStep, Data: state.Data, CreatedAt: state.CreatedAt, UpdatedAt: state.UpdatedAt, CompletedAt: state.CompletedAt, compensationExecuted: state.CompensationExecuted, executedOps: make(map[string]bool), } for _, stepState := range state.Steps { step := &SagaStep{ ID: stepState.ID, Name: stepState.Name, Status: stepState.Status, Data: stepState.Data, StartedAt: stepState.StartedAt, CompletedAt: stepState.CompletedAt, ExecutionID: stepState.ExecutionID, RetryCount: stepState.RetryCount, LastError: stepState.LastError, } saga.Steps = append(saga.Steps, step) if stepState.Status == "completed" { saga.executedOps[stepState.ExecutionID] = true } } return saga, nil } func (o *SagaOrchestrator) Stop() { close(o.stopChan) o.wg.Wait() for _, coord := range o.coordinators { close(coord.stopChan) coord.wg.Wait() } if o.logger != nil { o.logger.Info("Saga orchestrator stopped") } } // ============================================================================= // SAGA COORDINATOR // ============================================================================= func (c *SagaCoordinator) run() { defer c.wg.Done() interval := c.config.GetStuckCheckInterval() ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ticker.C: c.activeSagas.Range(func(key, value interface{}) bool { saga := value.(*SagaTransaction) if saga.Status == "running" { timeout := c.config.GetSagaTimeout() if time.Since(time.UnixMilli(saga.UpdatedAt)) > timeout { if err := c.orchestrator.resumeSagaFromStorage(saga.ID); err != nil { if c.orchestrator.logger != nil { c.orchestrator.logger.Error(fmt.Sprintf("Failed to recover stuck saga %s: %v", saga.ID, err)) } } } } return true }) case <-c.stopChan: return } } } func (o *SagaOrchestrator) resumeSagaFromStorage(sagaID string) error { state, err := o.storage.Load(sagaID) if err != nil { return err } if state == nil { return fmt.Errorf("saga %s not found in storage", sagaID) } if state.NodeID != "" && state.NodeID != o.nodeID { if o.isNodeAlive(state.NodeID) { return fmt.Errorf("saga %s is being executed on node %s", sagaID, state.NodeID) } } return o.resumeSaga(state) } // ============================================================================= // SAGA TRANSACTION // ============================================================================= type SagaStep struct { ID string `json:"id"` Name string `json:"name"` Execute func() error `json:"-"` Compensate func() error `json:"-"` Status string `json:"status"` Data map[string]interface{} `json:"data"` StartedAt int64 `json:"started_at"` CompletedAt int64 `json:"completed_at"` ExecutionID string `json:"execution_id"` RetryCount int `json:"retry_count"` LastError string `json:"last_error,omitempty"` } type SagaTransaction struct { ID string `json:"id"` Steps []*SagaStep `json:"steps"` CurrentStep int `json:"current_step"` Status string `json:"status"` CreatedAt int64 `json:"created_at"` UpdatedAt int64 `json:"updated_at"` CompletedAt int64 `json:"completed_at,omitempty"` Data map[string]interface{} `json:"data"` mu sync.RWMutex executedOps map[string]bool `json:"-"` compensationExecuted bool `json:"-"` } func (s *SagaTransaction) AddStep(name string, execute, compensate func() error, data map[string]interface{}) *SagaTransaction { s.mu.Lock() defer s.mu.Unlock() stepID := fmt.Sprintf("%s_step_%d_%d", s.ID, len(s.Steps), time.Now().UnixNano()) step := &SagaStep{ ID: stepID, Name: name, Execute: execute, Compensate: compensate, Status: "pending", Data: data, StartedAt: time.Now().UnixMilli(), ExecutionID: stepID, RetryCount: 0, } s.Steps = append(s.Steps, step) return s } func (s *SagaTransaction) SetData(key string, value interface{}) { s.mu.Lock() defer s.mu.Unlock() s.Data[key] = value } func (s *SagaTransaction) GetData(key string) (interface{}, bool) { s.mu.RLock() defer s.mu.RUnlock() val, ok := s.Data[key] return val, ok } func (s *SagaTransaction) HasExecuted(operationID string) bool { s.mu.RLock() defer s.mu.RUnlock() _, ok := s.executedOps[operationID] return ok } func (s *SagaTransaction) MarkExecuted(operationID string) { s.mu.Lock() defer s.mu.Unlock() s.executedOps[operationID] = true } // ============================================================================= // ГЛОБАЛЬНЫЕ SAGA // ============================================================================= var globalSagaOrchestrator *SagaOrchestrator var sagaOrchestratorMu sync.RWMutex func SetGlobalSagaOrchestrator(o *SagaOrchestrator) { sagaOrchestratorMu.Lock() defer sagaOrchestratorMu.Unlock() globalSagaOrchestrator = o } func GetGlobalSagaOrchestrator() *SagaOrchestrator { sagaOrchestratorMu.RLock() defer sagaOrchestratorMu.RUnlock() return globalSagaOrchestrator } // ============================================================================= // SAGA MANAGER - обёртка // ============================================================================= type SagaManager struct { orchestrator *SagaOrchestrator sagas sync.Map logger LoggerInterface stopChan chan struct{} wg sync.WaitGroup mu sync.RWMutex maxRetries int executedOps sync.Map } func NewSagaManager(logger LoggerInterface) *SagaManager { orchestrator, err := NewSagaOrchestrator(SagaStateDir, "default-node", logger) if err != nil { if logger != nil { logger.Error(fmt.Sprintf("Failed to create saga orchestrator: %v", err)) } return &SagaManager{ logger: logger, stopChan: make(chan struct{}), maxRetries: 3, } } sm := &SagaManager{ orchestrator: orchestrator, logger: logger, stopChan: make(chan struct{}), maxRetries: 3, } sm.wg.Add(1) go sm.cleanupExecutedOpsLoop() return sm } func (sm *SagaManager) BeginSaga(id string) *SagaTransaction { if sm.orchestrator != nil { saga, err := sm.orchestrator.BeginSaga(id) if err != nil { if sm.logger != nil { sm.logger.Error(fmt.Sprintf("Failed to begin saga: %v", err)) } return &SagaTransaction{ ID: id, Status: "pending", CreatedAt: time.Now().UnixMilli(), UpdatedAt: time.Now().UnixMilli(), Data: make(map[string]interface{}), executedOps: make(map[string]bool), } } sm.sagas.Store(id, saga) return saga } saga := &SagaTransaction{ ID: id, Steps: make([]*SagaStep, 0), CurrentStep: 0, Status: "pending", CreatedAt: time.Now().UnixMilli(), UpdatedAt: time.Now().UnixMilli(), Data: make(map[string]interface{}), executedOps: make(map[string]bool), } sm.sagas.Store(id, saga) LogTransactionAudit(TransactionID(0), "SAGA_BEGIN", TransactionActive, map[string]interface{}{ "saga_id": id, }) return saga } func (sm *SagaManager) Execute(saga *SagaTransaction) error { if sm.orchestrator != nil { return sm.orchestrator.Execute(saga) } saga.mu.Lock() defer saga.mu.Unlock() if saga.Status != "pending" { return fmt.Errorf("saga %s is not in pending state", saga.ID) } saga.Status = "running" saga.UpdatedAt = time.Now().UnixMilli() for i, step := range saga.Steps { saga.CurrentStep = i step.Status = "running" step.StartedAt = time.Now().UnixMilli() var err error for retry := 0; retry < sm.maxRetries; retry++ { step.RetryCount = retry + 1 if err = step.Execute(); err == nil { break } step.LastError = err.Error() time.Sleep(time.Duration(100*(retry+1)) * time.Millisecond) } step.CompletedAt = time.Now().UnixMilli() if err != nil { step.Status = "failed" saga.Status = "compensating" saga.UpdatedAt = time.Now().UnixMilli() compensationErr := sm.executeCompensation(saga, i) if compensationErr != nil { saga.Status = "compensation_failed" saga.UpdatedAt = time.Now().UnixMilli() LogTransactionAudit(TransactionID(0), "SAGA_COMPENSATION_FAILED", TransactionAborted, map[string]interface{}{ "saga_id": saga.ID, "failed_step": step.Name, "error": compensationErr.Error(), }) return fmt.Errorf("saga %s compensation failed: %v", saga.ID, compensationErr) } saga.Status = "aborted" saga.UpdatedAt = time.Now().UnixMilli() LogTransactionAudit(TransactionID(0), "SAGA_ABORTED", TransactionAborted, map[string]interface{}{ "saga_id": saga.ID, "failed_step": step.Name, "error": err.Error(), }) return fmt.Errorf("saga %s aborted at step %s: %v", saga.ID, step.Name, err) } step.Status = "completed" saga.MarkExecuted(step.ExecutionID) saga.UpdatedAt = time.Now().UnixMilli() } saga.Status = "completed" saga.UpdatedAt = time.Now().UnixMilli() saga.CompletedAt = saga.UpdatedAt LogTransactionAudit(TransactionID(0), "SAGA_COMPLETED", TransactionCommitted, map[string]interface{}{ "saga_id": saga.ID, "steps": len(saga.Steps), }) return nil } func (sm *SagaManager) executeCompensation(saga *SagaTransaction, failedStep int) error { if saga.compensationExecuted { return nil } for j := failedStep; j >= 0; j-- { step := saga.Steps[j] if step.Status == "compensated" || step.Status == "pending" { continue } var err error for retry := 0; retry < sm.maxRetries; retry++ { if err = step.Compensate(); err == nil { break } time.Sleep(time.Duration(100*(retry+1)) * time.Millisecond) } if err != nil { step.Status = "compensation_failed" return fmt.Errorf("compensation for step %s failed: %v", step.Name, err) } step.Status = "compensated" step.CompletedAt = time.Now().UnixMilli() } saga.compensationExecuted = true return nil } func (sm *SagaManager) cleanupExecutedOpsLoop() { defer sm.wg.Done() ticker := time.NewTicker(24 * time.Hour) defer ticker.Stop() cutoff := int64(7 * 24 * 3600 * 1000) for { select { case <-ticker.C: now := time.Now().UnixMilli() sm.executedOps.Range(func(key, value interface{}) bool { if ts, ok := value.(int64); ok && now-ts > cutoff { sm.executedOps.Delete(key) } return true }) case <-sm.stopChan: return } } } func (sm *SagaManager) GetSaga(id string) (*SagaTransaction, error) { if sm.orchestrator != nil { return sm.orchestrator.GetSaga(id) } if val, ok := sm.sagas.Load(id); ok { return val.(*SagaTransaction), nil } return nil, fmt.Errorf("saga %s not found", id) } func (sm *SagaManager) GetSagaStatus(id string) (string, error) { saga, err := sm.GetSaga(id) if err != nil { return "", err } saga.mu.RLock() defer saga.mu.RUnlock() return saga.Status, nil } func (sm *SagaManager) GetActiveSagas() []*SagaTransaction { result := make([]*SagaTransaction, 0) sm.sagas.Range(func(key, value interface{}) bool { saga := value.(*SagaTransaction) saga.mu.RLock() status := saga.Status saga.mu.RUnlock() if status == "pending" || status == "running" { result = append(result, saga) } return true }) return result } func (sm *SagaManager) GetOrchestrator() *SagaOrchestrator { return sm.orchestrator } func (sm *SagaManager) Stop() { close(sm.stopChan) sm.wg.Wait() if sm.orchestrator != nil { sm.orchestrator.Stop() } } // ============================================================================= // OPERATION, DocumentVersion // ============================================================================= type Operation struct { Type string `json:"type"` 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"` } type DocumentVersion struct { Document *Document `json:"document"` Timestamp int64 `json:"timestamp"` TxID TransactionID `json:"tx_id"` VersionID string `json:"version_id"` } // ============================================================================= // WAL MANAGER (не-сегментированный) // ============================================================================= 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 fsyncEnabled bool } func NewWALManager(path string, fsyncEnabled bool) (*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, fsyncEnabled: fsyncEnabled, } wm.wg.Add(1) go wm.writerLoop() return wm, nil } 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 } } } 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 } record.CRC = crc32(data) record.Length = uint32(len(data)) dataWithCRC, err := json.Marshal(record) if err != nil { continue } header := make([]byte, 8) binary.BigEndian.PutUint32(header[0:4], uint32(len(dataWithCRC))) binary.BigEndian.PutUint32(header[4:8], record.CRC) if _, err := wm.writer.Write(header); err != nil { continue } if _, err := wm.writer.Write(dataWithCRC); err != nil { continue } record.LSN = wm.currentLSN wm.currentLSN++ } } func (wm *WALManager) sync() { wm.mu.Lock() defer wm.mu.Unlock() if err := wm.writer.Flush(); err == nil { if wm.fsyncEnabled { if err := RealFsyncWithRetry(wm.file, FsyncMaxRetries, FsyncRetryDelay); err == nil { wm.lastSync = time.Now() } } else { wm.lastSync = time.Now() } } } 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") } } func (wm *WALManager) Sync() error { wm.mu.Lock() defer wm.mu.Unlock() if err := wm.writer.Flush(); err != nil { return err } if wm.fsyncEnabled { return RealFsyncWithRetry(wm.file, FsyncMaxRetries, FsyncRetryDelay) } return nil } 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) headerBuf := make([]byte, 8) for { _, err := io.ReadFull(reader, headerBuf) if err != nil { break } recordLen := binary.BigEndian.Uint32(headerBuf[0:4]) expectedCRC := binary.BigEndian.Uint32(headerBuf[4:8]) if recordLen == 0 || recordLen > 100*1024*1024 { break } recordData := make([]byte, recordLen) _, err = io.ReadFull(reader, recordData) if err != nil { break } var record WALRecord if err := json.Unmarshal(recordData, &record); err != nil { continue } data, _ := json.Marshal(record) calculatedCRC := crc32(data) if calculatedCRC != expectedCRC && calculatedCRC != record.CRC { continue } records = append(records, &record) } return records, nil } // Close - ИСПРАВЛЕНО: flush перед закрытием func (wm *WALManager) Close() error { wm.mu.Lock() if wm.closed { wm.mu.Unlock() return nil } wm.closed = true wm.mu.Unlock() close(wm.stopChan) wm.wg.Wait() close(wm.writeChan) wm.mu.Lock() defer wm.mu.Unlock() if err := wm.writer.Flush(); err != nil { return err } if wm.fsyncEnabled { RealFsync(wm.file) } return wm.file.Close() } // ============================================================================= // SEGMENTED WAL MANAGER // ============================================================================= type WALSegment struct { ID uint32 File *os.File Writer *bufio.Writer Path string StartLSN uint64 EndLSN uint64 Size int64 mu sync.Mutex } type WALIndexEntry struct { LSN uint64 SegmentID uint32 Offset int64 Length uint32 Checksum uint32 } type WALIndexManager struct { index map[uint64]*WALIndexEntry segments map[uint32]*WALSegment mu sync.RWMutex indexPath string } type SegmentedWALManager struct { segmentsDir string segments map[uint32]*WALSegment currentSegment *WALSegment currentSegmentID uint32 index *WALIndexManager mu sync.RWMutex writeChan chan *WALRecord stopCh chan struct{} ackChan chan uint64 // ИСПРАВЛЕНО: канал для подтверждения записи wg sync.WaitGroup batchSize int logger LoggerInterface recoveryManager *AsyncRecoveryManager recoveryComplete atomic.Bool backupLSN atomic.Uint64 fsyncEnabled bool } func NewSegmentedWALManager(segmentsDir string, fsyncEnabled bool, 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), stopCh: make(chan struct{}), ackChan: make(chan uint64, 10000), batchSize: 100, logger: logger, fsyncEnabled: fsyncEnabled, } 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 err := wm.validateSegments(); err != nil { if logger != nil { logger.Warn(fmt.Sprintf("WAL segment validation failed: %v", err)) } } if wm.currentSegment == nil { if err := wm.rotateSegmentInternal(); err != nil { return nil, err } } wm.wg.Add(1) go wm.writerLoop() return wm, nil } func (wm *SegmentedWALManager) validateSegments() error { wm.mu.Lock() defer wm.mu.Unlock() for id, segment := range wm.segments { if segment.File == nil { file, err := os.OpenFile(segment.Path, os.O_RDWR, 0644) if err != nil { if wm.logger != nil { wm.logger.Warn(fmt.Sprintf("WAL segment %d file missing: %s", id, segment.Path)) } delete(wm.segments, id) continue } segment.File = file segment.Writer = bufio.NewWriterSize(file, 64*1024) } } return nil } func (wm *SegmentedWALManager) GetBackupLSN() uint64 { return wm.backupLSN.Load() } func (wm *SegmentedWALManager) SetBackupLSN(lsn uint64) { wm.backupLSN.Store(lsn) } 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 } // rotateSegmentInternal - ИСПРАВЛЕНО: без захвата wm.mu (вызывается из flushBatch с уже захваченным mu) func (wm *SegmentedWALManager) rotateSegmentInternal() error { 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.getCurrentLSNLocked(), } if wm.currentSegment != nil { if err := wm.currentSegment.Writer.Flush(); err != nil { if wm.logger != nil { wm.logger.Error(fmt.Sprintf("Failed to flush old segment: %v", err)) } } if wm.fsyncEnabled { if err := RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay); err != nil { if wm.logger != nil { wm.logger.Error(fmt.Sprintf("Failed to fsync old segment: %v", err)) } } } wm.currentSegment.EndLSN = wm.getCurrentLSNLocked() wm.currentSegment.File.Close() FsyncDir(wm.segmentsDir) } 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 } // rotateSegment - публичный метод (захватывает mu) func (wm *SegmentedWALManager) rotateSegment() error { wm.mu.Lock() defer wm.mu.Unlock() return wm.rotateSegmentInternal() } func (wm *SegmentedWALManager) getCurrentLSNLocked() uint64 { if wm.currentSegment == nil { return 1 } return wm.currentSegment.StartLSN + uint64(wm.currentSegment.Size/100) } func (wm *SegmentedWALManager) getCurrentLSN() uint64 { wm.mu.RLock() defer wm.mu.RUnlock() return wm.getCurrentLSNLocked() } func (wm *SegmentedWALManager) Write(record *WALRecord) error { record.Timestamp = time.Now().UnixMilli() wm.writeChan <- record return nil } // WriteSync - ИСПРАВЛЕНО: синхронная запись с подтверждением через ackChan func (wm *SegmentedWALManager) WriteSync(record *WALRecord) error { record.Timestamp = time.Now().UnixMilli() wm.writeChan <- record // Ждём подтверждения записи (по LSN или таймауту) select { case <-wm.ackChan: return nil case <-time.After(5 * time.Second): return fmt.Errorf("WAL write sync timeout") } } 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 { 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] } case <-wm.stopCh: // ИСПРАВЛЕНО: flush оставшихся записей перед выходом if len(batch) > 0 { wm.flushBatch(batch) } return } } } // flushBatch - ИСПРАВЛЕНО: не освобождает wm.mu при ротации (вызывает rotateSegmentInternal) func (wm *SegmentedWALManager) flushBatch(batch []*WALRecord) { wm.mu.Lock() defer wm.mu.Unlock() for _, record := range batch { if wm.currentSegment.Size >= WALSegmentSize { // ИСПРАВЛЕНО: вызываем внутренний метод без повторного захвата mu if err := wm.rotateSegmentInternal(); err != nil { if wm.logger != nil { wm.logger.Error(fmt.Sprintf("Failed to rotate segment: %v", err)) } continue } } 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) header := make([]byte, 8) binary.BigEndian.PutUint32(header[0:4], uint32(len(data))) binary.BigEndian.PutUint32(header[4:8], record.CRC) if _, err := wm.currentSegment.Writer.Write(header); 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(8 + len(data)) wm.currentSegment.EndLSN = record.LSN } wm.currentSegment.Writer.Flush() if wm.fsyncEnabled { RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay) } wm.index.save() // ИСПРАВЛЕНО: отправляем подтверждение для всех записей в batch for range batch { select { case wm.ackChan <- 1: default: } } } func (wm *SegmentedWALManager) Sync() error { wm.mu.Lock() defer wm.mu.Unlock() if wm.currentSegment == nil { return nil } if err := wm.currentSegment.Writer.Flush(); err != nil { return err } if wm.fsyncEnabled { return RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay) } return nil } 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 { if wm.logger != nil { wm.logger.Warn(fmt.Sprintf("Failed to read segment %d: %v", seg.ID, err)) } continue } records = append(records, segRecords...) } return records, nil } func (wm *SegmentedWALManager) ReadSince(lsn uint64) ([]*WALRecord, error) { allRecords, err := wm.ReadAll() if err != nil { return nil, err } result := make([]*WALRecord, 0) for _, record := range allRecords { if record.LSN > lsn { result = append(result, record) } } return result, nil } func (wm *SegmentedWALManager) GetCurrentLSN() uint64 { wm.mu.RLock() defer wm.mu.RUnlock() if wm.currentSegment == nil { return 1 } return wm.currentSegment.EndLSN } 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) headerBuf := make([]byte, 8) for { _, err := io.ReadFull(reader, headerBuf) if err != nil { break } recordLen := binary.BigEndian.Uint32(headerBuf[0:4]) expectedCRC := binary.BigEndian.Uint32(headerBuf[4:8]) if recordLen == 0 || recordLen > 100*1024*1024 { break } recordData := make([]byte, recordLen) _, err = io.ReadFull(reader, 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...) calculatedCRC := crc32(crcData) if calculatedCRC != expectedCRC && calculatedCRC != record.CRC { continue } records = append(records, &record) } return records, nil } // Close - ИСПРАВЛЕНО: flush оставшихся записей перед закрытием func (wm *SegmentedWALManager) Close() error { close(wm.stopCh) wm.wg.Wait() close(wm.writeChan) wm.mu.Lock() defer wm.mu.Unlock() if wm.currentSegment != nil { wm.currentSegment.Writer.Flush() if wm.fsyncEnabled { RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay) } wm.currentSegment.File.Close() } wm.index.save() return nil } func (im *WALIndexManager) addEntry(entry *WALIndexEntry) { im.mu.Lock() defer im.mu.Unlock() im.index[entry.LSN] = entry } 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) } func (im *WALIndexManager) load() error { data, err := os.ReadFile(im.indexPath) if err != nil { if os.IsNotExist(err) { return nil } return err } if len(data) == 0 { return nil } return json.Unmarshal(data, &im.index) } // ============================================================================= // ASYNC RECOVERY MANAGER // ============================================================================= 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 } 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 } func (arm *AsyncRecoveryManager) worker() { defer arm.wg.Done() for record := range arm.recordChan { var err error if arm.callback != nil { err = arm.callback(record) } if err != nil { select { case arm.errChan <- err: default: } arm.errorCnt.Add(1) } else { arm.recoveredCnt.Add(1) } } } func (arm *AsyncRecoveryManager) errorMonitor() { criticalErrors := 0 for range arm.errChan { criticalErrors++ if criticalErrors > 10 { arm.Stop() return } } } 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 } } 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) arm.wg.Wait() close(arm.errChan) } 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(), } } // ============================================================================= // DEADLOCK DETECTOR // ============================================================================= type DeadlockDetector struct { waitForGraph sync.Map checkInterval time.Duration timeout time.Duration mu sync.RWMutex stopChan chan struct{} wg sync.WaitGroup logger LoggerInterface } 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 } func (dd *DeadlockDetector) SetLogger(logger LoggerInterface) { dd.logger = logger } func (dd *DeadlockDetector) Stop() { close(dd.stopChan) dd.wg.Wait() } 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) } func (dd *DeadlockDetector) RemoveWaiting(txID TransactionID) { dd.waitForGraph.Delete(txID) } 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 } } } 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 }) } func (dd *DeadlockDetector) resolveDeadlock(txID1, txID2 TransactionID) { if globalTxManager != nil { if val, ok := globalTxManager.activeTransactions.Load(txID1); ok { tx := val.(*Transaction) if tx.IsDistributed { if dd.logger != nil { dd.logger.Warn(fmt.Sprintf("Distributed transaction %d involved in deadlock with %d, aborting", txID1, txID2)) } AbortDistributedTransaction(txID1) } else { tx.State.Store(int32(TransactionAborted)) globalTxManager.activeTransactions.Delete(txID1) globalTxManager.stats.TotalDeadlocks.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) if dd.logger != nil { dd.logger.Warn(fmt.Sprintf("Deadlock resolved: aborted transaction %d due to conflict with %d", txID1, txID2)) } } } } } // ============================================================================= // DISTRIBUTED TX COORDINATOR // ============================================================================= type TxState int32 const ( TxActive TxState = iota TxCommitted TxAborted TxTimeout ) type DistributedTxInfo struct { TxID TransactionID Nodes []string Status TxState StartTime int64 Timeout time.Duration } type DistributedTransactionCoordinator struct { pendingTxs sync.Map timeout time.Duration mu sync.RWMutex } func NewDistributedTransactionCoordinator(timeout time.Duration) *DistributedTransactionCoordinator { if timeout <= 0 { timeout = 30 * time.Second } return &DistributedTransactionCoordinator{timeout: timeout} } func (dtc *DistributedTransactionCoordinator) Prepare(txID TransactionID, nodes []string) error { info := &DistributedTxInfo{ TxID: txID, Nodes: nodes, Status: TxActive, StartTime: time.Now().UnixMilli(), Timeout: dtc.timeout, } dtc.pendingTxs.Store(txID, info) return nil } 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 } func (dtc *DistributedTransactionCoordinator) Abort(txID TransactionID) error { dtc.pendingTxs.Delete(txID) return nil } // ============================================================================= // Transaction // ============================================================================= type Transaction struct { ID TransactionID State atomic.Int32 Operations []Operation StartTime int64 Version uint64 mu sync.RWMutex IsDistributed bool Nodes []string savepoints []*Savepoint timeout time.Duration timeoutTimer *time.Timer } type Savepoint struct { Name string Timestamp int64 OpCount int Snapshot *Document } type TransactionOptions struct { Timeout time.Duration IsDistributed bool Nodes []string IsolationLevel string } 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"` Savepoints []string `json:"savepoints,omitempty"` Nodes []string `json:"nodes,omitempty"` } type OperationInfo struct { Type string `json:"type"` Database string `json:"database"` Collection string `json:"collection"` DocumentID string `json:"document_id"` } type TransactionStats struct { TotalStarted atomic.Uint64 TotalCommitted atomic.Uint64 TotalAborted atomic.Uint64 TotalTimedOut atomic.Uint64 TotalDeadlocks atomic.Uint64 ActiveCount atomic.Uint64 PeakActiveCount atomic.Uint64 MaxOpsPerTx atomic.Uint64 AvgOpsPerTx atomic.Uint64 TotalOps atomic.Uint64 StartTime time.Time } type TransactionManager struct { activeTransactions sync.Map nextTxID atomic.Uint64 wal *SegmentedWALManager logger LoggerInterface mu sync.RWMutex walPath string checkpointInterval int64 lastCheckpoint int64 checkpointFile *os.File documentVersions sync.Map maxVersions int visibilityMap *VisibilityMap readCache *ReadTimestampCache distCoord *DistributedTransactionCoordinator deadlockDetector *DeadlockDetector recoveryManager *AsyncRecoveryManager recoveryComplete atomic.Bool backupLock sync.RWMutex backupInProgress atomic.Bool stats *TransactionStats mvccManager *MVCCManager } // ============================================================================= // ГЛОБАЛЬНЫЕ ПЕРЕМЕННЫЕ // ============================================================================= var ( globalTxManager *TransactionManager txManagerOnce sync.Once currentTx atomic.Value globalStorage *Storage ) func InitTransactionManager(walPath string) error { return InitTransactionManagerWithConfig(walPath, nil) } 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(30 * time.Second), deadlockDetector: NewDeadlockDetector(DeadlockCheckInterval, DefaultTxTimeout), stats: &TransactionStats{StartTime: time.Now()}, mvccManager: NewMVCCManager(MaxVersionsPerDoc, VersionRetentionDays), } globalTxManager.nextTxID.Store(1) var walErr error fsyncEnabled := true if config != nil { if v, ok := config["fsync_enabled"].(bool); ok { fsyncEnabled = v } } globalTxManager.wal, walErr = NewSegmentedWALManager(filepath.Dir(walPath), fsyncEnabled, nil) if walErr != nil { err = walErr return } globalTxManager.startAsyncRecovery() go globalTxManager.checkpointLoop() go globalTxManager.versionCleanupLoop() go globalTxManager.statsMonitor() }) return err } func GetTransactionManager() *TransactionManager { return globalTxManager } 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) } } } func SetGlobalStorage(s *Storage) { globalStorage = s } func GetGlobalStorage() *Storage { return globalStorage } func BeginTransaction() *Transaction { if globalTxManager == nil { InitTransactionManager("futriis.wal") } return BeginTransactionWithOptions(&TransactionOptions{ Timeout: DefaultTxTimeout, }) } func BeginTransactionWithOptions(options *TransactionOptions) *Transaction { if globalTxManager == nil { InitTransactionManager("futriis.wal") } if options == nil { options = &TransactionOptions{Timeout: DefaultTxTimeout} } tx := &Transaction{ ID: TransactionID(globalTxManager.nextTxID.Add(1) - 1), StartTime: time.Now().UnixMilli(), Operations: make([]Operation, 0, 100), Version: 1, savepoints: make([]*Savepoint, 0), timeout: options.Timeout, IsDistributed: options.IsDistributed, Nodes: options.Nodes, } tx.State.Store(int32(TransactionActive)) globalTxManager.activeTransactions.Store(tx.ID, tx) currentTx.Store(tx) globalTxManager.stats.TotalStarted.Add(1) globalTxManager.stats.ActiveCount.Add(1) LogTransactionAudit(tx.ID, "START", TransactionActive, map[string]interface{}{ "start_time": tx.StartTime, "timeout_ms": options.Timeout.Milliseconds(), "distributed": options.IsDistributed, }) if options.Timeout > 0 { tx.timeoutTimer = time.AfterFunc(options.Timeout, func() { if TransactionState(tx.State.Load()) == TransactionActive { tx.State.Store(int32(TransactionAborted)) globalTxManager.activeTransactions.Delete(tx.ID) globalTxManager.stats.TotalTimedOut.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) LogTransactionAudit(tx.ID, "TIMEOUT", TransactionAborted, map[string]interface{}{ "timeout_ms": options.Timeout.Milliseconds(), }) } }) } return tx } func BeginTransactionWithTimeout(timeout time.Duration) *Transaction { return BeginTransactionWithOptions(&TransactionOptions{Timeout: timeout}) } func BeginDistributedTransaction(nodes []string) (*Transaction, error) { if globalTxManager == nil { if err := InitTransactionManager("futriis.wal"); err != nil { return nil, err } } options := &TransactionOptions{ Timeout: 30 * time.Second, IsDistributed: true, Nodes: nodes, IsolationLevel: "READ_COMMITTED", } tx := BeginTransactionWithOptions(options) if tx == nil { return nil, fmt.Errorf("failed to create transaction") } if err := globalTxManager.distCoord.Prepare(tx.ID, nodes); err != nil { return nil, err } if globalTxManager.logger != nil { globalTxManager.logger.Info(fmt.Sprintf("Distributed transaction %d started on nodes: %v", tx.ID, nodes)) } return tx, nil } // CommitCurrentTransaction - ИСПРАВЛЕНО: сначала WAL, потом применение 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") } if tx.IsDistributed { return fmt.Errorf("distributed transaction must use CommitDistributedTransaction") } if tx.timeout > 0 && time.Since(time.UnixMilli(tx.StartTime)) > tx.timeout { AbortCurrentTransaction() return fmt.Errorf("transaction timeout exceeded") } // ИСПРАВЛЕНО: Сначала записываем транзакцию в WAL (prepared state) txRecord := &TransactionRecord{ ID: tx.ID, State: TransactionCommitted, Timestamp: time.Now().UnixMilli(), Operations: tx.Operations, } data, err := json.Marshal(txRecord) if err != nil { return fmt.Errorf("failed to marshal transaction: %v", err) } walRecord := &WALRecord{Type: 1, Data: data} // ИСПРАВЛЕНО: используем WriteSync для гарантии записи в WAL до применения if err := globalTxManager.wal.WriteSync(walRecord); err != nil { return fmt.Errorf("failed to write transaction to WAL: %v", err) } // Теперь применяем операции 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 { if globalTxManager.mvccManager != nil { globalTxManager.mvccManager.CreateVersion(doc, tx.ID) } globalTxManager.AddDocumentVersion(op.DocumentID, &DocumentVersion{ Document: doc.Clone(), Timestamp: time.Now().UnixMilli(), TxID: tx.ID, }) } } } } } } tx.State.Store(int32(TransactionCommitted)) LogTransactionAudit(tx.ID, "COMMIT", TransactionCommitted, map[string]interface{}{ "operations": len(tx.Operations), }) globalTxManager.stats.TotalCommitted.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) globalTxManager.stats.TotalOps.Add(uint64(len(tx.Operations))) if uint64(len(tx.Operations)) > globalTxManager.stats.MaxOpsPerTx.Load() { globalTxManager.stats.MaxOpsPerTx.Store(uint64(len(tx.Operations))) } if tx.timeoutTimer != nil { tx.timeoutTimer.Stop() } currentTx.Store(nil) globalTxManager.activeTransactions.Delete(tx.ID) return nil } func AbortCurrentTransaction() error { txVal := currentTx.Load() if txVal == nil { return fmt.Errorf("no active transaction") } tx := txVal.(*Transaction) tx.State.Store(int32(TransactionAborted)) LogTransactionAudit(tx.ID, "ABORT", TransactionAborted, map[string]interface{}{ "operations": len(tx.Operations), }) globalTxManager.stats.TotalAborted.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) if tx.timeoutTimer != nil { tx.timeoutTimer.Stop() } currentTx.Store(nil) globalTxManager.activeTransactions.Delete(tx.ID) return nil } func CommitDistributedTransaction(txID TransactionID) error { if globalTxManager == nil { return fmt.Errorf("transaction manager not initialized") } val, ok := globalTxManager.activeTransactions.Load(txID) if !ok { return fmt.Errorf("transaction not found: %d", txID) } tx := val.(*Transaction) if TransactionState(tx.State.Load()) != TransactionActive { return fmt.Errorf("transaction is not active") } // ИСПРАВЛЕНО: Сначала WAL txRecord := &TransactionRecord{ ID: tx.ID, State: TransactionCommitted, Timestamp: time.Now().UnixMilli(), Operations: tx.Operations, IsDistributed: true, Nodes: tx.Nodes, } data, err := json.Marshal(txRecord) if err != nil { return fmt.Errorf("failed to marshal: %v", err) } if err := globalTxManager.wal.WriteSync(&WALRecord{Type: 1, Data: data}); err != nil { return fmt.Errorf("failed to write to WAL: %v", err) } if err := globalTxManager.distCoord.Commit(txID); err != nil { return err } for _, op := range tx.Operations { if err := applyOperation(op); err != nil { return fmt.Errorf("failed to apply operation: %v", err) } } tx.State.Store(int32(TransactionCommitted)) globalTxManager.activeTransactions.Delete(txID) globalTxManager.stats.TotalCommitted.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) if tx.timeoutTimer != nil { tx.timeoutTimer.Stop() } LogTransactionAudit(txID, "COMMIT_DISTRIBUTED", TransactionCommitted, map[string]interface{}{ "nodes": tx.Nodes, }) return nil } func AbortDistributedTransaction(txID TransactionID) error { if globalTxManager == nil { return fmt.Errorf("transaction manager not initialized") } val, ok := globalTxManager.activeTransactions.Load(txID) if !ok { return fmt.Errorf("transaction not found: %d", txID) } tx := val.(*Transaction) if err := globalTxManager.distCoord.Abort(txID); err != nil { return err } tx.State.Store(int32(TransactionAborted)) globalTxManager.activeTransactions.Delete(txID) globalTxManager.stats.TotalAborted.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) if tx.timeoutTimer != nil { tx.timeoutTimer.Stop() } LogTransactionAudit(txID, "ABORT_DISTRIBUTED", TransactionAborted, map[string]interface{}{ "nodes": tx.Nodes, }) return nil } 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 if globalTxManager != nil && globalTxManager.mvccManager != nil { globalTxManager.mvccManager.CreateVersion(doc, 0) } return coll.Insert(doc) case "update": if err := coll.Update(op.DocumentID, op.Data); err != nil { return err } if globalTxManager != nil && globalTxManager.mvccManager != nil { if doc, err := coll.Find(op.DocumentID); err == nil { globalTxManager.mvccManager.CreateVersion(doc, 0) } } return nil case "delete": return coll.Delete(op.DocumentID) case "restore": return coll.RestoreDeleted(op.DocumentID) } return nil } func (tx *Transaction) CreateSavepoint(name string) error { if TransactionState(tx.State.Load()) != TransactionActive { return fmt.Errorf("transaction is not active") } if len(tx.savepoints) >= MaxSavepointsPerTx { return fmt.Errorf("too many savepoints (max %d)", MaxSavepointsPerTx) } for _, sp := range tx.savepoints { if sp.Name == name { return fmt.Errorf("savepoint '%s' already exists", name) } } savepoint := &Savepoint{ Name: name, Timestamp: time.Now().UnixMilli(), OpCount: len(tx.Operations), } if len(tx.Operations) > 0 { lastOp := tx.Operations[len(tx.Operations)-1] if lastOp.DocumentID != "" && globalStorage != nil { db, _ := globalStorage.GetDatabase(lastOp.Database) if db != nil { coll, _ := db.GetCollection(lastOp.Collection) if coll != nil { if doc, err := coll.Find(lastOp.DocumentID); err == nil { savepoint.Snapshot = doc.Clone() } } } } } tx.mu.Lock() tx.savepoints = append(tx.savepoints, savepoint) tx.mu.Unlock() LogTransactionAudit(tx.ID, "SAVEPOINT", TransactionActive, map[string]interface{}{ "savepoint": name, "op_count": savepoint.OpCount, }) return nil } func (tx *Transaction) RollbackToSavepoint(name string) error { if TransactionState(tx.State.Load()) != TransactionActive { return fmt.Errorf("transaction is not active") } tx.mu.Lock() defer tx.mu.Unlock() var targetIdx int = -1 for i, sp := range tx.savepoints { if sp.Name == name { targetIdx = i break } } if targetIdx == -1 { return fmt.Errorf("savepoint '%s' not found", name) } if len(tx.Operations) > tx.savepoints[targetIdx].OpCount { tx.Operations = tx.Operations[:tx.savepoints[targetIdx].OpCount] } tx.savepoints = tx.savepoints[:targetIdx+1] LogTransactionAudit(tx.ID, "ROLLBACK_TO_SAVEPOINT", TransactionActive, map[string]interface{}{ "savepoint": name, }) return nil } func (tx *Transaction) ReleaseSavepoint(name string) error { tx.mu.Lock() defer tx.mu.Unlock() for i, sp := range tx.savepoints { if sp.Name == name { tx.savepoints = append(tx.savepoints[:i], tx.savepoints[i+1:]...) LogTransactionAudit(tx.ID, "RELEASE_SAVEPOINT", TransactionActive, map[string]interface{}{ "savepoint": name, }) return nil } } return fmt.Errorf("savepoint '%s' not found", name) } func (tx *Transaction) GetSavepoints() []string { tx.mu.RLock() defer tx.mu.RUnlock() names := make([]string, len(tx.savepoints)) for i, sp := range tx.savepoints { names[i] = sp.Name } return names } 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) } } 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 } func (tm *TransactionManager) checkpointLoop() { ticker := time.NewTicker(time.Duration(tm.checkpointInterval) * time.Second) defer ticker.Stop() for range ticker.C { tm.createCheckpoint() } } // createCheckpoint - ИСПРАВЛЕНО: fsync для чекпоинта 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["backup_lsn"] = tm.wal.GetBackupLSN() data, err := json.Marshal(checkpoint) if err != nil { if tm.logger != nil { tm.logger.Error(fmt.Sprintf("Failed to marshal checkpoint: %v", err)) } return } // ИСПРАВЛЕНО: используем writeFileSync с fsync if err := writeFileSync(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)) } } 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 }) } } func (tm *TransactionManager) statsMonitor() { ticker := time.NewTicker(10 * time.Second) defer ticker.Stop() for range ticker.C { active := tm.stats.ActiveCount.Load() if active > tm.stats.PeakActiveCount.Load() { tm.stats.PeakActiveCount.Store(active) } } } 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 records recovered", stats["recovered"])) } 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 } } }() } func (tm *TransactionManager) IsRecoveryComplete() bool { return tm.recoveryComplete.Load() } 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"], } } func (tm *TransactionManager) LockForBackup() { tm.backupLock.Lock() tm.backupInProgress.Store(true) } func (tm *TransactionManager) UnlockForBackup() { tm.backupInProgress.Store(false) tm.backupLock.Unlock() } func (tm *TransactionManager) IsBackupInProgress() bool { return tm.backupInProgress.Load() } func GetTransactionStats() map[string]interface{} { if globalTxManager == nil { return map[string]interface{}{"error": "transaction manager not initialized"} } stats := globalTxManager.stats active := stats.ActiveCount.Load() totalStarted := stats.TotalStarted.Load() totalCommitted := stats.TotalCommitted.Load() totalAborted := stats.TotalAborted.Load() return map[string]interface{}{ "total_started": totalStarted, "total_committed": totalCommitted, "total_aborted": totalAborted, "total_timed_out": stats.TotalTimedOut.Load(), "total_deadlocks": stats.TotalDeadlocks.Load(), "active_count": active, "peak_active_count": stats.PeakActiveCount.Load(), "max_ops_per_tx": stats.MaxOpsPerTx.Load(), "avg_ops_per_tx": stats.AvgOpsPerTx.Load(), "total_ops": stats.TotalOps.Load(), "commit_rate": float64(totalCommitted) / float64(totalStarted+1) * 100, "abort_rate": float64(totalAborted) / float64(totalStarted+1) * 100, "uptime_seconds": time.Since(stats.StartTime).Seconds(), "is_recovery_complete": globalTxManager.IsRecoveryComplete(), "backup_in_progress": globalTxManager.IsBackupInProgress(), } } func StopTransactionManager() error { if globalTxManager == nil { return nil } if globalTxManager.deadlockDetector != nil { globalTxManager.deadlockDetector.Stop() } if globalTxManager.wal != nil { globalTxManager.wal.Sync() return globalTxManager.wal.Close() } return nil } func HasActiveTransaction() bool { return currentTx.Load() != nil } func GetCurrentTransactionID() string { txVal := currentTx.Load() if txVal == nil { return "" } tx := txVal.(*Transaction) return fmt.Sprintf("%d", tx.ID) } 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" state := TransactionState(tx.State.Load()) switch state { case TransactionCommitted: status = "committed" case 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, }) } savepoints := tx.GetSavepoints() tx.mu.RUnlock() info := TransactionInfo{ ID: fmt.Sprintf("%d", tx.ID), Status: status, StartTime: tx.StartTime, OperationCount: opCount, Operations: operations, Savepoints: savepoints, } if tx.IsDistributed { info.Status = "distributed_" + status info.Nodes = tx.Nodes } transactions = append(transactions, info) return true }) return transactions } 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") } 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() LogTransactionAudit(tx.ID, "ADD_OPERATION", TransactionActive, map[string]interface{}{ "operation": opType, "document": doc.ID, }) return nil } 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) } func MVCCSnapshot() uint64 { return uint64(time.Now().UnixNano()) } func CreateDocumentVersion(doc *Document, txID TransactionID) *DocumentVersion { return &DocumentVersion{ Document: doc.Clone(), Timestamp: time.Now().UnixMilli(), TxID: txID, } } 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 } func CheckTransactionTimeout(txID TransactionID) error { if globalTxManager == nil { return fmt.Errorf("transaction manager not initialized") } val, ok := globalTxManager.activeTransactions.Load(txID) if !ok { return fmt.Errorf("transaction not found: %d", txID) } tx := val.(*Transaction) if TransactionState(tx.State.Load()) != TransactionActive { return nil } if tx.timeout > 0 && time.Since(time.UnixMilli(tx.StartTime)) > tx.timeout { tx.State.Store(int32(TransactionAborted)) globalTxManager.activeTransactions.Delete(tx.ID) globalTxManager.stats.TotalTimedOut.Add(1) globalTxManager.stats.ActiveCount.Add(^uint64(0)) if tx.timeoutTimer != nil { tx.timeoutTimer.Stop() } LogTransactionAudit(txID, "TIMEOUT_CHECK", TransactionAborted, map[string]interface{}{ "timeout_ms": tx.timeout.Milliseconds(), "elapsed_ms": time.Since(time.UnixMilli(tx.StartTime)).Milliseconds(), }) return fmt.Errorf("transaction %d timed out", txID) } return nil } // ============================================================================= // ФУНКЦИИ МИГРАЦИИ // ============================================================================= func GetDocumentAtTimestamp(docID string, timestamp int64) (*Document, error) { if globalTxManager == nil { return nil, fmt.Errorf("transaction manager not initialized") } doc := globalTxManager.GetDocumentVersion(docID, timestamp) if doc == nil { return nil, fmt.Errorf("document %s not found at timestamp %d", docID, timestamp) } return doc, nil } func GetDocumentVersionsSince(docID string, since int64) ([]*DocumentVersion, error) { if globalTxManager == nil { return nil, fmt.Errorf("transaction manager not initialized") } val, ok := globalTxManager.documentVersions.Load(docID) if !ok { return nil, fmt.Errorf("no versions found for document %s", docID) } versions := val.([]*DocumentVersion) result := make([]*DocumentVersion, 0) for _, v := range versions { if v.Timestamp > since { result = append(result, v) } } return result, nil } func CreateSnapshot(database, collection string) (map[string][]*DocumentVersion, error) { if globalStorage == nil { return nil, fmt.Errorf("storage not initialized") } db, err := globalStorage.GetDatabase(database) if err != nil { return nil, err } coll, err := db.GetCollection(collection) if err != nil { return nil, err } docs := coll.GetAllDocuments() result := make(map[string][]*DocumentVersion) for _, doc := range docs { versions, err := GetDocumentVersionsSince(doc.ID, 0) if err != nil { continue } if len(versions) > 0 { result[doc.ID] = versions } else if globalTxManager != nil { version := &DocumentVersion{ Document: doc.Clone(), Timestamp: time.Now().UnixMilli(), TxID: TransactionID(0), VersionID: fmt.Sprintf("%s_%d", doc.ID, time.Now().UnixNano()), } globalTxManager.AddDocumentVersion(doc.ID, version) result[doc.ID] = []*DocumentVersion{version} } } return result, nil } func ApplySnapshot(database, collection string, snapshot map[string][]*DocumentVersion) error { if globalStorage == nil { return fmt.Errorf("storage not initialized") } db, err := globalStorage.GetDatabase(database) if err != nil { return err } coll, err := db.GetCollection(collection) if err != nil { return err } for docID, versions := range snapshot { if len(versions) == 0 { continue } latest := versions[len(versions)-1] doc := latest.Document existing, err := coll.Find(docID) if err == nil && existing != nil { updates := doc.GetFields() if err := coll.Update(docID, updates); err != nil { return fmt.Errorf("failed to update document %s: %v", docID, err) } } else { if err := coll.Insert(doc); err != nil { return fmt.Errorf("failed to insert document %s: %v", docID, err) } } if globalTxManager != nil { for _, v := range versions { globalTxManager.AddDocumentVersion(docID, v) } } } return nil } func GetMVCCManager() *MVCCManager { if globalTxManager == nil { return nil } return globalTxManager.mvccManager } func GetMVCCStats() map[string]interface{} { if globalTxManager == nil || globalTxManager.mvccManager == nil { return map[string]interface{}{"error": "MVCC manager not initialized"} } return globalTxManager.mvccManager.GetMVCCStats() } func GetVisibilityMapStats() map[string]interface{} { if globalTxManager == nil || globalTxManager.visibilityMap == nil { return map[string]interface{}{"error": "Visibility map not initialized"} } return globalTxManager.visibilityMap.GetStats() } func GetReadCacheStats() map[string]interface{} { if globalTxManager == nil || globalTxManager.readCache == nil { return map[string]interface{}{"error": "Read cache not initialized"} } return globalTxManager.readCache.GetStats() } // ============================================================================= // ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ // ============================================================================= func maxUint64(a, b uint64) uint64 { if a > b { return a } return b }