Files
futriix/internal/storage/transactions.go
T

4122 lines
131 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/*
* 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<<retry)) * time.Millisecond
if o.config != nil {
backoff = time.Duration(o.config.RetryBackoffMs*(1<<retry)) * time.Millisecond
}
time.Sleep(backoff)
}
step.CompletedAt = time.Now().UnixMilli()
if err != nil {
step.Status = "failed"
saga.Status = "compensating"
saga.UpdatedAt = time.Now().UnixMilli()
o.saveSagaState(saga)
compensationErr := o.executeCompensation(saga, i)
if compensationErr != nil {
saga.Status = "compensation_failed"
saga.UpdatedAt = time.Now().UnixMilli()
o.saveSagaState(saga)
o.metrics.TotalFailed.Add(1)
o.metrics.ActiveCount.Add(-1)
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()
o.saveSagaState(saga)
o.metrics.TotalAborted.Add(1)
o.metrics.ActiveCount.Add(-1)
duration := time.Since(startTime).Milliseconds()
o.recordDuration(duration)
LogTransactionAudit(TransactionID(0), "SAGA_ABORTED", TransactionAborted, map[string]interface{}{
"saga_id": saga.ID,
"failed_step": step.Name,
"error": err.Error(),
"duration_ms": duration,
})
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()
o.saveSagaState(saga)
}
saga.Status = "completed"
saga.CompletedAt = time.Now().UnixMilli()
saga.UpdatedAt = saga.CompletedAt
o.saveSagaState(saga)
o.metrics.TotalCompleted.Add(1)
o.metrics.ActiveCount.Add(-1)
duration := time.Since(startTime).Milliseconds()
o.recordDuration(duration)
LogTransactionAudit(TransactionID(0), "SAGA_COMPLETED", TransactionCommitted, map[string]interface{}{
"saga_id": saga.ID,
"steps": len(saga.Steps),
"duration_ms": duration,
})
if o.logger != nil {
o.logger.Info(fmt.Sprintf("Saga %s completed successfully in %dms", saga.ID, duration))
}
return nil
}
func (o *SagaOrchestrator) saveSagaState(saga *SagaTransaction) error {
state := &SagaState{
ID: saga.ID,
Status: saga.Status,
CurrentStep: saga.CurrentStep,
Data: saga.Data,
CreatedAt: saga.CreatedAt,
UpdatedAt: saga.UpdatedAt,
CompletedAt: saga.CompletedAt,
CompensationExecuted: saga.compensationExecuted,
NodeID: o.nodeID,
CoordinatorID: o.coordinators[0].id,
ExecutionID: fmt.Sprintf("%s_%d", saga.ID, saga.CreatedAt),
}
state.Steps = make([]SagaStepState, len(saga.Steps))
for i, step := range saga.Steps {
state.Steps[i] = SagaStepState{
ID: step.ID,
Name: step.Name,
Status: step.Status,
Data: step.Data,
StartedAt: step.StartedAt,
CompletedAt: step.CompletedAt,
ExecutionID: step.ExecutionID,
RetryCount: step.RetryCount,
LastError: step.LastError,
}
}
return o.storage.Save(state)
}
func (o *SagaOrchestrator) 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
}
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<<retry)) * time.Millisecond
if o.config != nil {
backoff = time.Duration(o.config.RetryBackoffMs*(1<<retry)) * time.Millisecond
}
time.Sleep(backoff)
}
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()
o.saveSagaState(saga)
}
}
saga.compensationExecuted = true
o.metrics.TotalCompensated.Add(1)
return nil
}
func (o *SagaOrchestrator) recordDuration(durationMs int64) {
o.metrics.mu.Lock()
defer o.metrics.mu.Unlock()
o.metrics.TotalDurationMs.Add(uint64(durationMs))
avg := o.metrics.TotalDurationMs.Load() / maxUint64(1, o.metrics.TotalCompleted.Load())
o.metrics.AvgDurationMs.Store(avg)
if o.metrics.minLatency == -1 || durationMs < o.metrics.minLatency {
o.metrics.minLatency = durationMs
}
if durationMs > 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
}