gvsafronov/futriix
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
c26037bd24fbdd342340bb984d218ff3414473d5
futriix/internal/storage/transactions.go

3204 lines
99 KiB
Go
Raw Normal View History

Upload files to "internal/storage"
2026-07-14 21:14:27 +00:00
/*
* 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 (Multi-Version Concurrency Control) и WAL (Write-Ahead Log) без блокировок.
// Для распределённых транзакций используется протокол SAGA
package storage
import (
"bufio"
"encoding/binary"
"encoding/json"
"fmt"
"os"
"path/filepath"
"sort"
"sync"
"sync/atomic"
"time"
)
// =============================================================================
// БАЗОВЫЕ ТИПЫ
// =============================================================================
// TransactionID - уникальный идентификатор транзакции.
type TransactionID uint64
// TransactionState - состояние транзакции.
type TransactionState int32
const (
TransactionActive TransactionState = iota
TransactionCommitted
TransactionAborted
TransactionPrepared
)
// TransactionRecord - запись транзакции в WAL.
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"`
}
// WALRecord - запись в WAL (Write-Ahead Log).
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_"
VisibilityMapSize = 1024 * 1024
VersionPruneInterval = 5 * time.Minute
MaxVersionsPerDoc = 100
VersionRetentionDays = 7
DefaultTxTimeout = 30 * time.Second
DeadlockCheckInterval = 1 * time.Second
MaxSavepointsPerTx = 100
AsyncRecoveryBufferSize = 10000
AsyncRecoveryWorkers = 4
AsyncRecoveryTimeout = 30 * time.Second
FsyncMaxRetries = 3
FsyncRetryDelay = 100 * time.Millisecond
)
// =============================================================================
// CRC32 - КОНТРОЛЬНАЯ СУММА
// =============================================================================
var crc32Table = [256]uint32{
0x00000000, 0x77073096, 0xee0e612c, 0x990951ba, 0x076dc419, 0x706af48f,
0xe963a535, 0x9e6495a3, 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988,
0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91, 0x1db71064, 0x6ab020f2,
0xf3b97148, 0x84be41de, 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7,
0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec, 0x14015c4f, 0x63066cd9,
0xfa0f3d63, 0x8d080df5, 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172,
0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b, 0x35b5a8fa, 0x42b2986c,
0xdbbbc9d6, 0xacbcf940, 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59,
0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116, 0x21b4f4b5, 0x56b3c423,
0xcfba9599, 0xb8bda50f, 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924,
0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d, 0x76dc4190, 0x01db7106,
0x98d220bc, 0xefd5102a, 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433,
0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818, 0x7f6a0dbb, 0x086d3d2d,
0x91646c97, 0xe6635c01, 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e,
0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457, 0x65b0d9c6, 0x12b7e950,
0x8bbeb8ea, 0xfcb9887c, 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65,
0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2, 0x4adfa541, 0x3dd895d7,
0xa4d1c46d, 0xd3d6f4fb, 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0,
0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9, 0x5005713c, 0x270241aa,
0xbe0b1010, 0xc90c2086, 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f,
0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4, 0x59b33d17, 0x2eb40d81,
0xb7bd5c3b, 0xc0ba6cad, 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a,
0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683, 0xe3630b12, 0x94643b84,
0x0d6d6a3e, 0x7a6a5aa8, 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1,
0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe, 0xf762575d, 0x806567cb,
0x196c3671, 0x6e6b06e7, 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc,
0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5, 0xd6d6a3e8, 0xa1d1937e,
0x38d8c2c4, 0x4fdff252, 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b,
0xd80d2bda, 0xaf0a1a4c, 0x36034af6, 0x41047a60, 0xdf60efc3, 0xa867df55,
0x316e8eef, 0x4669be79, 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236,
0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f, 0xc5ba3bbe, 0xb2bd0b28,
0x2bb45a92, 0x5cb36a04, 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d,
0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a, 0x9c0906a9, 0xeb0e363f,
0x72076785, 0x05005713, 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38,
0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21, 0x86d3d2d4, 0xf1d4e242,
0x68ddb3f8, 0x1fda836e, 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777,
0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c, 0x8f659eff, 0xf862ae69,
0x616bffd3, 0x166ccf45, 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2,
0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db, 0xaed16a4a, 0xd9d65adc,
0x40df0b66, 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
}
// =============================================================================
// MVCC - MULTI-VERSION CONCURRENCY CONTROL (ПОЛНАЯ РЕАЛИЗАЦИЯ)
// =============================================================================
// MVCCManager - основной менеджер MVCC для production использования
type MVCCManager struct {
// Версии документов: docID -> []*DocumentVersion
versions sync.Map
// Карта видимости для быстрой проверки
visibilityMap *VisibilityMap
// Кэш для чтения по временной метке
readCache *ReadTimestampCache
// Настройки
maxVersionsPerDoc int
retentionDuration time.Duration
// Статистика
stats MVCCStats
// Защита для операций с версиями
mu sync.RWMutex
}
// MVCCStats статистика MVCC
type MVCCStats struct {
TotalVersionsCreated atomic.Uint64
TotalVersionsPruned atomic.Uint64
TotalCacheHits atomic.Uint64
TotalCacheMisses atomic.Uint64
TotalVisibilityHits atomic.Uint64
TotalVisibilityMisses atomic.Uint64
}
// NewMVCCManager создаёт новый MVCC менеджер
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),
}
// Запускаем фоновую очистку старых версий
go m.pruneOldVersionsLoop()
return m
}
// CreateVersion создаёт новую версию документа
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 {
// Удаляем самые старые версии, но сохраняем хотя бы одну
versions = versions[len(versions)-m.maxVersionsPerDoc:]
m.stats.TotalVersionsPruned.Add(uint64(len(versions)))
}
m.versions.Store(doc.ID, versions)
// Отмечаем версию как видимую
m.visibilityMap.MarkVisible(doc.ID, uint64(txID), true)
m.stats.TotalVersionsCreated.Add(1)
return version
}
// GetVersionAt возвращает версию документа на указанный момент времени
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
}
// GetLatestVersion возвращает последнюю версию документа
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()
}
// GetAllVersions возвращает все версии документа
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
}
// pruneOldVersionsLoop периодически удаляет старые версии
func (m *MVCCManager) pruneOldVersionsLoop() {
ticker := time.NewTicker(VersionPruneInterval)
defer ticker.Stop()
for range ticker.C {
m.PruneOldVersions()
}
}
// PruneOldVersions удаляет старые версии документов
func (m *MVCCManager) PruneOldVersions() {
cutoffTime := time.Now().Add(-m.retentionDuration).UnixMilli()
pruned := int64(0)
m.mu.Lock()
defer m.mu.Unlock()
m.versions.Range(func(key, value interface{}) bool {
docID := key.(string)
versions := value.([]*DocumentVersion)
// Оставляем только версии новее cutoff
newVersions := make([]*DocumentVersion, 0, len(versions))
for _, v := range versions {
if v.Timestamp >= cutoffTime {
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))
}
}
// GetMVCCStats возвращает статистику MVCC
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(),
"max_versions_per_doc": m.maxVersionsPerDoc,
"retention_days": int(m.retentionDuration.Hours() / 24),
}
}
// =============================================================================
// VISIBILITY MAP
// =============================================================================
// VisibilityMapEntry - запись в карте видимости.
type VisibilityMapEntry struct {
DocID string
VisibleFrom uint64
VisibleTo uint64
IsVisible bool
LastAccess int64
}
// VisibilityMap - карта видимости версий.
type VisibilityMap struct {
entries sync.Map
maxSize int
hitCount atomic.Uint64
missCount atomic.Uint64
mu sync.RWMutex
}
// NewVisibilityMap создаёт новую карту видимости.
func NewVisibilityMap(maxSize int) *VisibilityMap {
if maxSize <= 0 {
maxSize = VisibilityMapSize
}
return &VisibilityMap{maxSize: maxSize}
}
// MarkVisible - отмечает версию как видимую.
func (vm *VisibilityMap) MarkVisible(docID string, version uint64, visible bool) {
key := fmt.Sprintf("%s@%d", docID, version)
vm.entries.Store(key, &VisibilityMapEntry{
DocID: docID,
VisibleFrom: version,
VisibleTo: version,
IsVisible: visible,
LastAccess: time.Now().Unix(),
})
}
// IsVisible - проверяет, видима ли версия.
func (vm *VisibilityMap) IsVisible(docID string, version uint64) bool {
key := fmt.Sprintf("%s@%d", docID, version)
// Сначала проверяем точное совпадение
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
}
// GetStats - возвращает статистику карты видимости.
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
// =============================================================================
// ReadTimestampCache - кэш для чтения по временной метке.
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
}
// NewReadTimestampCache создаёт новый кэш чтения.
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
}
// Get - получает документ из кэша.
func (rtc *ReadTimestampCache) Get(docID string, timestamp int64) *Document {
key := fmt.Sprintf("%s@%d", docID, timestamp)
val, ok := rtc.cache.Load(key)
if !ok {
rtc.misses.Add(1)
return nil
}
entry := val.(*cachedEntry)
if time.Since(entry.cachedAt) > rtc.ttl {
rtc.cache.Delete(key)
rtc.size.Add(-1)
rtc.misses.Add(1)
return nil
}
entry.accessCount++
rtc.hits.Add(1)
return entry.doc
}
// Set - сохраняет документ в кэше.
func (rtc *ReadTimestampCache) Set(docID string, timestamp int64, doc *Document) {
key := fmt.Sprintf("%s@%d", docID, timestamp)
// Проверяем размер кэша
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)
}
// evictOldest удаляет самый старый элемент кэша
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)
}
}
// cleanupLoop периодически очищает устаревшие записи
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
})
}
}
// GetStats - возвращает статистику кэша.
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 - УЛУЧШЕННАЯ ВЕРСИЯ С ИДЕМПОТЕНТНОСТЬЮ
// =============================================================================
// SagaStep представляет шаг в Saga транзакции.
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"` // Уникальный ID выполнения для идемпотентности
RetryCount int `json:"retry_count"`
LastError string `json:"last_error,omitempty"`
}
// SagaTransaction представляет Saga транзакцию.
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"`
Data map[string]interface{} `json:"data"`
mu sync.RWMutex
// История выполненных операций для идемпотентности
executedOps map[string]bool `json:"-"`
compensationExecuted bool `json:"-"`
}
// SagaManager управляет Saga транзакциями.
type SagaManager struct {
sagas sync.Map
logger LoggerInterface
stopChan chan struct{}
wg sync.WaitGroup
mu sync.RWMutex
maxRetries int
// Хранилище выполненных операций для идемпотентности (глобальное)
executedOps sync.Map
}
// NewSagaManager создаёт новый менеджер Saga.
func NewSagaManager(logger LoggerInterface) *SagaManager {
sm := &SagaManager{
logger: logger,
stopChan: make(chan struct{}),
maxRetries: 3,
}
// Запускаем фоновую очистку старых записей
go sm.cleanupExecutedOpsLoop()
return sm
}
// BeginSaga начинает новую Saga транзакцию.
func (sm *SagaManager) BeginSaga(id string) *SagaTransaction {
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),
compensationExecuted: false,
}
sm.sagas.Store(id, saga)
LogAudit("SAGA_BEGIN", "SAGA", id, map[string]interface{}{
"created_at": saga.CreatedAt,
})
return saga
}
// AddStep добавляет шаг в Saga транзакцию.
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, // Уникальный ID выполнения для идемпотентности
RetryCount: 0,
}
s.Steps = append(s.Steps, step)
return s
}
// SetData устанавливает данные в Saga транзакции.
func (s *SagaTransaction) SetData(key string, value interface{}) {
s.mu.Lock()
defer s.mu.Unlock()
s.Data[key] = value
}
// GetData получает данные из Saga транзакции.
func (s *SagaTransaction) GetData(key string) (interface{}, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
val, ok := s.Data[key]
return val, ok
}
// HasExecuted проверяет, была ли выполнена операция (идемпотентность)
func (s *SagaTransaction) HasExecuted(operationID string) bool {
s.mu.RLock()
defer s.mu.RUnlock()
_, ok := s.executedOps[operationID]
return ok
}
// MarkExecuted отмечает операцию как выполненную
func (s *SagaTransaction) MarkExecuted(operationID string) {
s.mu.Lock()
defer s.mu.Unlock()
s.executedOps[operationID] = true
}
// Execute выполняет Saga транзакцию (оркестратор).
func (sm *SagaManager) Execute(saga *SagaTransaction) error {
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()
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Executing saga step %s: %s (execution_id: %s)",
saga.ID, step.Name, step.ExecutionID))
}
// Проверка идемпотентности (глобальная)
if sm.isOperationExecuted(step.ExecutionID) {
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Saga step %s already executed globally (idempotent), skipping", step.Name))
}
step.Status = "completed"
step.CompletedAt = time.Now().UnixMilli()
continue
}
// Проверка локальной идемпотентности
if saga.HasExecuted(step.ExecutionID) {
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Saga step %s already executed locally (idempotent), skipping", step.Name))
}
step.Status = "completed"
step.CompletedAt = time.Now().UnixMilli()
continue
}
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()
if sm.logger != nil {
sm.logger.Warn(fmt.Sprintf("Saga step %s failed (attempt %d/%d): %v",
step.Name, retry+1, sm.maxRetries, err))
}
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()
if sm.logger != nil {
sm.logger.Error(fmt.Sprintf("Saga step %s failed, starting compensation", step.Name))
}
// Компенсация с идемпотентностью
compensationErr := sm.executeCompensation(saga, i)
if compensationErr != nil {
saga.Status = "compensation_failed"
saga.UpdatedAt = time.Now().UnixMilli()
if sm.logger != nil {
sm.logger.Error(fmt.Sprintf("Compensation for saga %s failed: %v", saga.ID, compensationErr))
}
LogAudit("SAGA_COMPENSATION_FAILED", "SAGA", saga.ID, map[string]interface{}{
"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()
LogAudit("SAGA_ABORTED", "SAGA", saga.ID, map[string]interface{}{
"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"
sm.markOperationExecuted(step.ExecutionID)
saga.MarkExecuted(step.ExecutionID)
saga.UpdatedAt = time.Now().UnixMilli()
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Saga step %s completed successfully", step.Name))
}
}
saga.Status = "completed"
saga.UpdatedAt = time.Now().UnixMilli()
if sm.logger != nil {
sm.logger.Info(fmt.Sprintf("Saga %s completed successfully", saga.ID))
}
LogAudit("SAGA_COMPLETED", "SAGA", saga.ID, map[string]interface{}{
"steps": len(saga.Steps),
})
return nil
}
// executeCompensation выполняет компенсацию с идемпотентностью
func (sm *SagaManager) executeCompensation(saga *SagaTransaction, failedStep int) error {
if saga.compensationExecuted {
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Compensation for saga %s already executed (idempotent)", saga.ID))
}
return nil
}
// Компенсируем шаги в обратном порядке
for j := failedStep; j >= 0; j-- {
step := saga.Steps[j]
if step.Status == "compensated" || step.Status == "pending" {
continue
}
// Проверка идемпотентности компенсации
compensationID := step.ExecutionID + "_compensate"
if sm.isOperationExecuted(compensationID) {
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Compensation for step %s already executed globally (idempotent), skipping", step.Name))
}
step.Status = "compensated"
continue
}
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Compensating step %s", step.Name))
}
var err error
for retry := 0; retry < sm.maxRetries; retry++ {
if err = step.Compensate(); err == nil {
break
}
if sm.logger != nil {
sm.logger.Warn(fmt.Sprintf("Compensation for step %s failed (attempt %d/%d): %v",
step.Name, retry+1, sm.maxRetries, err))
}
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"
sm.markOperationExecuted(compensationID)
if sm.logger != nil {
sm.logger.Debug(fmt.Sprintf("Compensation for step %s completed", step.Name))
}
}
saga.compensationExecuted = true
return nil
}
// isOperationExecuted проверяет, была ли выполнена операция (глобальная идемпотентность)
func (sm *SagaManager) isOperationExecuted(operationID string) bool {
_, ok := sm.executedOps.Load(operationID)
return ok
}
// markOperationExecuted отмечает операцию как выполненную
func (sm *SagaManager) markOperationExecuted(operationID string) {
sm.executedOps.Store(operationID, time.Now().UnixMilli())
}
// cleanupExecutedOpsLoop очищает старые записи выполненных операций
func (sm *SagaManager) cleanupExecutedOpsLoop() {
ticker := time.NewTicker(24 * time.Hour)
defer ticker.Stop()
cutoff := int64(7 * 24 * 3600 * 1000) // 7 дней в миллисекундах
for range ticker.C {
now := time.Now().UnixMilli()
sm.executedOps.Range(func(key, value interface{}) bool {
if ts, ok := value.(int64); ok {
if now-ts > cutoff {
sm.executedOps.Delete(key)
}
}
return true
})
}
}
// GetSaga возвращает Saga по ID.
func (sm *SagaManager) GetSaga(id string) (*SagaTransaction, error) {
if val, ok := sm.sagas.Load(id); ok {
return val.(*SagaTransaction), nil
}
return nil, fmt.Errorf("saga %s not found", id)
}
// GetSagaStatus возвращает статус Saga.
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
}
// GetActiveSagas возвращает все активные Saga транзакции.
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
}
// Stop останавливает менеджер Saga.
func (sm *SagaManager) Stop() {
close(sm.stopChan)
sm.wg.Wait()
}
// =============================================================================
// ОСТАЛЬНЫЕ ТИПЫ
// =============================================================================
// Operation - операция транзакции.
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"`
}
// DocumentVersion - версия документа.
type DocumentVersion struct {
Document *Document `json:"document"`
Timestamp int64 `json:"timestamp"`
TxID TransactionID `json:"tx_id"`
VersionID string `json:"version_id"`
}
// =============================================================================
// WAL MANAGER - УПРАВЛЕНИЕ ЖУРНАЛОМ ПРЕДЗАПИСИ
// =============================================================================
// WALManager управляет WAL (Write-Ahead Log).
type WALManager struct {
mu sync.RWMutex
file *os.File
writer *bufio.Writer
path string
currentLSN uint64
lastSync time.Time
syncInterval time.Duration
bufferSize int
closed bool
writeChan chan *WALRecord
stopChan chan struct{}
wg sync.WaitGroup
batchSize int
fsyncEnabled bool
}
// NewWALManager создаёт новый WAL менеджер.
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
}
// writerLoop - основной цикл асинхронной записи.
func (wm *WALManager) writerLoop() {
defer wm.wg.Done()
batch := make([]*WALRecord, 0, wm.batchSize)
ticker := time.NewTicker(wm.syncInterval)
defer ticker.Stop()
for {
select {
case record, ok := <-wm.writeChan:
if !ok {
if len(batch) > 0 {
wm.flushBatch(batch)
}
return
}
batch = append(batch, record)
if len(batch) >= wm.batchSize {
wm.flushBatch(batch)
batch = batch[:0]
}
case <-ticker.C:
if len(batch) > 0 {
wm.flushBatch(batch)
batch = batch[:0]
}
if time.Since(wm.lastSync) >= wm.syncInterval {
wm.sync()
}
case <-wm.stopChan:
if len(batch) > 0 {
wm.flushBatch(batch)
}
wm.sync()
return
}
}
}
// flushBatch - записывает пакет записей в WAL.
func (wm *WALManager) flushBatch(batch []*WALRecord) {
wm.mu.Lock()
defer wm.mu.Unlock()
for _, record := range batch {
data, err := json.Marshal(record)
if err != nil {
continue
}
lenBuf := make([]byte, 4)
binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
if _, err := wm.writer.Write(lenBuf); err != nil {
continue
}
if _, err := wm.writer.Write(data); err != nil {
continue
}
record.LSN = wm.currentLSN
wm.currentLSN++
}
}
// sync - синхронизирует данные с диском.
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()
}
}
}
// Write - записывает запись в WAL (асинхронно).
func (wm *WALManager) Write(record *WALRecord) error {
if wm.closed {
return fmt.Errorf("WAL is closed")
}
record.Timestamp = time.Now().UnixMilli()
data, err := json.Marshal(record.Data)
if err != nil {
return err
}
record.CRC = crc32(data)
select {
case wm.writeChan <- record:
return nil
case <-time.After(100 * time.Millisecond):
return fmt.Errorf("WAL write timeout")
}
}
// Sync - принудительная синхронизация WAL с диском
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
}
// ReadAll - читает все записи из WAL.
func (wm *WALManager) ReadAll() ([]*WALRecord, error) {
wm.mu.RLock()
defer wm.mu.RUnlock()
wm.writer.Flush()
file, err := os.Open(wm.path)
if err != nil {
return nil, err
}
defer file.Close()
records := make([]*WALRecord, 0)
reader := bufio.NewReader(file)
lenBuf := make([]byte, 4)
for {
_, err := reader.Read(lenBuf)
if err != nil {
break
}
recordLen := binary.BigEndian.Uint32(lenBuf)
recordData := make([]byte, recordLen)
_, err = reader.Read(recordData)
if err != nil {
break
}
var record WALRecord
if err := json.Unmarshal(recordData, &record); err != nil {
continue
}
data, _ := json.Marshal(record.Data)
if crc32(data) != record.CRC {
continue
}
records = append(records, &record)
}
return records, nil
}
// Close - закрывает WAL менеджер.
func (wm *WALManager) Close() error {
wm.mu.Lock()
wm.closed = true
wm.mu.Unlock()
close(wm.stopChan)
close(wm.writeChan)
wm.wg.Wait()
wm.mu.Lock()
defer wm.mu.Unlock()
if err := wm.writer.Flush(); err != nil {
return err
}
if wm.fsyncEnabled {
RealFsync(wm.file)
}
return wm.file.Close()
}
// =============================================================================
// SEGMENTED WAL MANAGER - СЕГМЕНТИРОВАННЫЙ ЖУРНАЛ
// =============================================================================
// WALSegment - сегмент WAL.
type WALSegment struct {
ID uint32
File *os.File
Writer *bufio.Writer
Path string
StartLSN uint64
EndLSN uint64
Size int64
mu sync.Mutex
}
// WALIndexEntry - запись в индексе WAL.
type WALIndexEntry struct {
LSN uint64
SegmentID uint32
Offset int64
Length uint32
Checksum uint32
}
// WALIndexManager - управление индексом WAL.
type WALIndexManager struct {
index map[uint64]*WALIndexEntry
segments map[uint32]*WALSegment
mu sync.RWMutex
indexPath string
}
// SegmentedWALManager - менеджер сегментированного WAL.
type SegmentedWALManager struct {
segmentsDir string
segments map[uint32]*WALSegment
currentSegment *WALSegment
currentSegmentID uint32
index *WALIndexManager
mu sync.RWMutex
writeChan chan *WALRecord
stopChan chan struct{}
wg sync.WaitGroup
batchSize int
logger LoggerInterface
recoveryManager *AsyncRecoveryManager
recoveryComplete atomic.Bool
backupLSN atomic.Uint64
fsyncEnabled bool
}
// NewSegmentedWALManager создаёт новый сегментированный WAL менеджер.
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),
stopChan: make(chan struct{}),
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 wm.currentSegment == nil {
if err := wm.rotateSegment(); err != nil {
return nil, err
}
}
wm.wg.Add(1)
go wm.writerLoop()
return wm, nil
}
// GetBackupLSN - возвращает LSN для бэкапа.
func (wm *SegmentedWALManager) GetBackupLSN() uint64 {
return wm.backupLSN.Load()
}
// SetBackupLSN - устанавливает LSN для бэкапа.
func (wm *SegmentedWALManager) SetBackupLSN(lsn uint64) {
wm.backupLSN.Store(lsn)
}
// loadExistingSegments - загружает существующие сегменты из директории.
func (wm *SegmentedWALManager) loadExistingSegments() error {
files, err := filepath.Glob(filepath.Join(wm.segmentsDir, WALSegmentPrefix+"*"))
if err != nil {
return err
}
for _, filePath := range files {
var segmentID uint32
if _, err := fmt.Sscanf(filepath.Base(filePath), WALSegmentPrefix+"%d.log", &segmentID); err != nil {
continue
}
file, err := os.OpenFile(filePath, os.O_RDWR, 0644)
if err != nil {
continue
}
stat, _ := file.Stat()
segment := &WALSegment{
ID: segmentID,
File: file,
Writer: bufio.NewWriterSize(file, 64*1024),
Path: filePath,
Size: stat.Size(),
StartLSN: uint64(segmentID) * WALSegmentSize / 100,
}
wm.segments[segmentID] = segment
if segmentID > wm.currentSegmentID {
wm.currentSegmentID = segmentID
wm.currentSegment = segment
}
}
return nil
}
// rotateSegment - создаёт новый сегмент.
func (wm *SegmentedWALManager) rotateSegment() error {
wm.mu.Lock()
defer wm.mu.Unlock()
newSegmentID := wm.currentSegmentID + 1
segmentPath := filepath.Join(wm.segmentsDir, fmt.Sprintf(WALSegmentPrefix+"%d.log", newSegmentID))
file, err := os.OpenFile(segmentPath, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0644)
if err != nil {
return fmt.Errorf("failed to create segment: %v", err)
}
newSegment := &WALSegment{
ID: newSegmentID,
File: file,
Writer: bufio.NewWriterSize(file, 64*1024),
Path: segmentPath,
StartLSN: wm.getCurrentLSN(),
}
if wm.currentSegment != nil {
wm.currentSegment.Writer.Flush()
if wm.fsyncEnabled {
RealFsync(wm.currentSegment.File)
}
wm.currentSegment.File.Close()
}
wm.currentSegment = newSegment
wm.currentSegmentID = newSegmentID
wm.segments[newSegmentID] = newSegment
if wm.logger != nil {
wm.logger.Info(fmt.Sprintf("Created new WAL segment: %d", newSegmentID))
}
return nil
}
// getCurrentLSN - возвращает текущий LSN.
func (wm *SegmentedWALManager) getCurrentLSN() uint64 {
wm.mu.RLock()
defer wm.mu.RUnlock()
if wm.currentSegment == nil {
return 1
}
return wm.currentSegment.StartLSN + uint64(wm.currentSegment.Size/100)
}
// Write - записывает запись в WAL.
func (wm *SegmentedWALManager) Write(record *WALRecord) error {
record.Timestamp = time.Now().UnixMilli()
wm.writeChan <- record
return nil
}
// writerLoop - основной цикл асинхронной записи.
func (wm *SegmentedWALManager) writerLoop() {
defer wm.wg.Done()
batch := make([]*WALRecord, 0, wm.batchSize)
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case record, ok := <-wm.writeChan:
if !ok {
wm.flushBatch(batch)
return
}
batch = append(batch, record)
if len(batch) >= wm.batchSize {
wm.flushBatch(batch)
batch = batch[:0]
}
case <-ticker.C:
if len(batch) > 0 {
wm.flushBatch(batch)
batch = batch[:0]
}
case <-wm.stopChan:
wm.flushBatch(batch)
return
}
}
}
// flushBatch - записывает пакет записей в текущий сегмент.
func (wm *SegmentedWALManager) flushBatch(batch []*WALRecord) {
wm.mu.Lock()
defer wm.mu.Unlock()
for _, record := range batch {
if wm.currentSegment.Size >= WALSegmentSize {
wm.mu.Unlock()
wm.rotateSegment()
wm.mu.Lock()
}
data, err := json.Marshal(record)
if err != nil {
continue
}
lsnBytes := make([]byte, 8)
binary.BigEndian.PutUint64(lsnBytes, record.LSN)
crcData := append(lsnBytes, data...)
record.CRC = crc32(crcData)
lenBuf := make([]byte, 4)
binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
if _, err := wm.currentSegment.Writer.Write(lenBuf); err != nil {
continue
}
if _, err := wm.currentSegment.Writer.Write(data); err != nil {
continue
}
wm.index.addEntry(&WALIndexEntry{
LSN: record.LSN,
SegmentID: wm.currentSegment.ID,
Offset: wm.currentSegment.Size,
Length: uint32(len(data)),
Checksum: record.CRC,
})
wm.currentSegment.Size += int64(4 + len(data))
wm.currentSegment.EndLSN = record.LSN
}
wm.currentSegment.Writer.Flush()
if wm.fsyncEnabled {
RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay)
}
wm.index.save()
}
// Sync - принудительная синхронизация WAL с диском
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
}
// ReadAll - читает все записи из всех сегментов.
func (wm *SegmentedWALManager) ReadAll() ([]*WALRecord, error) {
wm.mu.RLock()
segments := make([]*WALSegment, 0, len(wm.segments))
for _, seg := range wm.segments {
segments = append(segments, seg)
}
wm.mu.RUnlock()
sort.Slice(segments, func(i, j int) bool {
return segments[i].ID < segments[j].ID
})
records := make([]*WALRecord, 0)
for _, seg := range segments {
segRecords, err := wm.readSegmentRecords(seg)
if err != nil {
return nil, err
}
records = append(records, segRecords...)
}
return records, nil
}
// ReadSince - читает записи начиная с указанного LSN.
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
}
// GetCurrentLSN - возвращает текущий LSN.
func (wm *SegmentedWALManager) GetCurrentLSN() uint64 {
wm.mu.RLock()
defer wm.mu.RUnlock()
if wm.currentSegment == nil {
return 1
}
return wm.currentSegment.EndLSN
}
// readSegmentRecords - читает записи из конкретного сегмента.
func (wm *SegmentedWALManager) readSegmentRecords(seg *WALSegment) ([]*WALRecord, error) {
seg.mu.Lock()
defer seg.mu.Unlock()
if seg.File == nil {
return nil, nil
}
seg.Writer.Flush()
seg.File.Seek(0, 0)
records := make([]*WALRecord, 0)
reader := bufio.NewReader(seg.File)
lenBuf := make([]byte, 4)
for {
_, err := reader.Read(lenBuf)
if err != nil {
break
}
recordLen := binary.BigEndian.Uint32(lenBuf)
recordData := make([]byte, recordLen)
_, err = reader.Read(recordData)
if err != nil {
break
}
var record WALRecord
if err := json.Unmarshal(recordData, &record); err != nil {
continue
}
lsnBytes := make([]byte, 8)
binary.BigEndian.PutUint64(lsnBytes, record.LSN)
crcData := append(lsnBytes, recordData...)
if crc32(crcData) != record.CRC {
continue
}
records = append(records, &record)
}
return records, nil
}
// Close - закрывает WAL менеджер.
func (wm *SegmentedWALManager) Close() error {
close(wm.stopChan)
close(wm.writeChan)
wm.wg.Wait()
wm.mu.Lock()
defer wm.mu.Unlock()
if wm.currentSegment != nil {
wm.currentSegment.Writer.Flush()
if wm.fsyncEnabled {
RealFsyncWithRetry(wm.currentSegment.File, FsyncMaxRetries, FsyncRetryDelay)
}
wm.currentSegment.File.Close()
}
wm.index.save()
return nil
}
// addEntry - добавляет запись в индекс.
func (im *WALIndexManager) addEntry(entry *WALIndexEntry) {
im.mu.Lock()
defer im.mu.Unlock()
im.index[entry.LSN] = entry
}
// save - сохраняет индекс на диск.
func (im *WALIndexManager) save() error {
im.mu.RLock()
defer im.mu.RUnlock()
data, err := json.Marshal(im.index)
if err != nil {
return err
}
return os.WriteFile(im.indexPath, data, 0644)
}
// load - загружает индекс с диска.
func (im *WALIndexManager) load() error {
data, err := os.ReadFile(im.indexPath)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return err
}
if len(data) == 0 {
return nil
}
return json.Unmarshal(data, &im.index)
}
// =============================================================================
// ASYNC RECOVERY MANAGER - АСИНХРОННОЕ ВОССТАНОВЛЕНИЕ
// =============================================================================
// AsyncRecoveryManager - управляет асинхронным восстановлением.
type AsyncRecoveryManager struct {
recordChan chan *WALRecord
errChan chan error
doneChan chan struct{}
wg sync.WaitGroup
callback func(*WALRecord) error
mu sync.RWMutex
isRunning bool
recoveredCnt atomic.Uint64
errorCnt atomic.Uint64
startTime time.Time
}
// NewAsyncRecoveryManager создаёт новый менеджер асинхронного восстановления.
func NewAsyncRecoveryManager(callback func(*WALRecord) error, workers int) *AsyncRecoveryManager {
arm := &AsyncRecoveryManager{
recordChan: make(chan *WALRecord, AsyncRecoveryBufferSize),
errChan: make(chan error, workers),
doneChan: make(chan struct{}),
callback: callback,
startTime: time.Now(),
isRunning: true,
}
for i := 0; i < workers; i++ {
arm.wg.Add(1)
go arm.worker()
}
go arm.errorMonitor()
return arm
}
// worker - воркер для обработки записей восстановления.
func (arm *AsyncRecoveryManager) worker() {
defer arm.wg.Done()
for record := range arm.recordChan {
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)
}
}
}
// errorMonitor - мониторинг ошибок восстановления.
func (arm *AsyncRecoveryManager) errorMonitor() {
criticalErrors := 0
for range arm.errChan {
criticalErrors++
if criticalErrors > 10 {
arm.Stop()
return
}
}
}
// Push - отправляет запись на восстановление.
func (arm *AsyncRecoveryManager) Push(record *WALRecord) bool {
arm.mu.RLock()
if !arm.isRunning {
arm.mu.RUnlock()
return false
}
arm.mu.RUnlock()
select {
case arm.recordChan <- record:
return true
case <-time.After(100 * time.Millisecond):
return false
}
}
// Wait - ожидает завершения восстановления.
func (arm *AsyncRecoveryManager) Wait() {
close(arm.recordChan)
arm.wg.Wait()
close(arm.doneChan)
}
// Stop - останавливает восстановление.
func (arm *AsyncRecoveryManager) Stop() {
arm.mu.Lock()
if !arm.isRunning {
arm.mu.Unlock()
return
}
arm.isRunning = false
arm.mu.Unlock()
close(arm.recordChan)
}
// GetStats - возвращает статистику восстановления.
func (arm *AsyncRecoveryManager) GetStats() map[string]interface{} {
return map[string]interface{}{
"recovered": arm.recoveredCnt.Load(),
"errors": arm.errorCnt.Load(),
"is_running": arm.isRunning,
"elapsed_ms": time.Since(arm.startTime).Milliseconds(),
}
}
// =============================================================================
// DEADLOCK DETECTOR - ОБНАРУЖЕНИЕ ВЗАИМНЫХ БЛОКИРОВОК
// =============================================================================
// DeadlockDetector - детектор взаимных блокировок.
type DeadlockDetector struct {
waitForGraph sync.Map
checkInterval time.Duration
timeout time.Duration
mu sync.RWMutex
stopChan chan struct{}
wg sync.WaitGroup
logger LoggerInterface
}
// NewDeadlockDetector - создаёт новый детектор дедлоков.
func NewDeadlockDetector(checkInterval, timeout time.Duration) *DeadlockDetector {
if checkInterval <= 0 {
checkInterval = DeadlockCheckInterval
}
if timeout <= 0 {
timeout = DefaultTxTimeout
}
d := &DeadlockDetector{
checkInterval: checkInterval,
timeout: timeout,
stopChan: make(chan struct{}),
}
d.wg.Add(1)
go d.detectLoop()
return d
}
// SetLogger - устанавливает логгер.
func (dd *DeadlockDetector) SetLogger(logger LoggerInterface) {
dd.logger = logger
}
// Stop - останавливает детектор.
func (dd *DeadlockDetector) Stop() {
close(dd.stopChan)
dd.wg.Wait()
}
// AddWaiting - добавляет зависимость ожидания.
func (dd *DeadlockDetector) AddWaiting(waiting, waitingFor TransactionID) {
var list []TransactionID
if val, ok := dd.waitForGraph.Load(waiting); ok {
list = val.([]TransactionID)
}
list = append(list, waitingFor)
dd.waitForGraph.Store(waiting, list)
}
// RemoveWaiting - удаляет зависимость ожидания.
func (dd *DeadlockDetector) RemoveWaiting(txID TransactionID) {
dd.waitForGraph.Delete(txID)
}
// detectLoop - основной цикл обнаружения дедлоков.
func (dd *DeadlockDetector) detectLoop() {
defer dd.wg.Done()
ticker := time.NewTicker(dd.checkInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
dd.detect()
case <-dd.stopChan:
return
}
}
}
// detect - выполняет обнаружение дедлоков.
func (dd *DeadlockDetector) detect() {
visited := make(map[TransactionID]bool)
stack := make(map[TransactionID]bool)
var dfs func(txID TransactionID) bool
dfs = func(txID TransactionID) bool {
visited[txID] = true
stack[txID] = true
val, ok := dd.waitForGraph.Load(txID)
if ok {
for _, next := range val.([]TransactionID) {
if !visited[next] {
if dfs(next) {
return true
}
} else if stack[next] {
dd.resolveDeadlock(txID, next)
return true
}
}
}
stack[txID] = false
return false
}
dd.waitForGraph.Range(func(key, value interface{}) bool {
txID := key.(TransactionID)
if !visited[txID] {
dfs(txID)
}
return true
})
}
// resolveDeadlock - разрешает взаимную блокировку.
func (dd *DeadlockDetector) resolveDeadlock(txID1, txID2 TransactionID) {
if globalTxManager != nil {
if val, ok := globalTxManager.activeTransactions.Load(txID1); ok {
tx := val.(*Transaction)
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 TRANSACTION COORDINATOR
// =============================================================================
// TxState - состояние распределённой транзакции.
type TxState int32
const (
TxActive TxState = iota
TxCommitted
TxAborted
TxTimeout
)
// DistributedTxInfo - информация о распределённой транзакции.
type DistributedTxInfo struct {
TxID TransactionID
Nodes []string
Status TxState
StartTime int64
Timeout time.Duration
}
// DistributedTransactionCoordinator - координатор распределённых транзакций.
type DistributedTransactionCoordinator struct {
pendingTxs sync.Map
timeout time.Duration
mu sync.RWMutex
}
// NewDistributedTransactionCoordinator - создаёт новый координатор.
func NewDistributedTransactionCoordinator(timeout time.Duration) *DistributedTransactionCoordinator {
if timeout <= 0 {
timeout = 30 * time.Second
}
return &DistributedTransactionCoordinator{timeout: timeout}
}
// Prepare - подготавливает распределённую транзакцию.
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
}
// Commit - коммитит распределённую транзакцию.
func (dtc *DistributedTransactionCoordinator) Commit(txID TransactionID) error {
val, ok := dtc.pendingTxs.Load(txID)
if !ok {
return fmt.Errorf("transaction not found: %d", txID)
}
info := val.(*DistributedTxInfo)
info.Status = TxCommitted
return nil
}
// Abort - отменяет распределённую транзакцию.
func (dtc *DistributedTransactionCoordinator) Abort(txID TransactionID) error {
dtc.pendingTxs.Delete(txID)
return nil
}
// =============================================================================
// Transaction - структура транзакции.
// =============================================================================
// 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
}
// Savepoint - точка сохранения в транзакции.
type Savepoint struct {
Name string
Timestamp int64
OpCount int
Snapshot *Document
}
// TransactionOptions - опции создания транзакции.
type TransactionOptions struct {
Timeout time.Duration
IsDistributed bool
Nodes []string
IsolationLevel string
}
// TransactionInfo - информация о транзакции.
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"`
}
// OperationInfo - информация об операции.
type OperationInfo struct {
Type string `json:"type"`
Database string `json:"database"`
Collection string `json:"collection"`
DocumentID string `json:"document_id"`
}
// TransactionStats - статистика по транзакциям.
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
}
// TransactionManager - основной менеджер транзакций.
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
// MVCC менеджер
mvccManager *MVCCManager
}
// =============================================================================
// ГЛОБАЛЬНЫЕ ПЕРЕМЕННЫЕ И ФУНКЦИИ ДОСТУПА
// =============================================================================
var (
globalTxManager *TransactionManager
txManagerOnce sync.Once
currentTx atomic.Value
globalStorage *Storage
)
// InitTransactionManager - инициализирует менеджер транзакций.
func InitTransactionManager(walPath string) error {
return InitTransactionManagerWithConfig(walPath, nil)
}
// InitTransactionManagerWithConfig - инициализирует менеджер транзакций с конфигурацией.
func InitTransactionManagerWithConfig(walPath string, config map[string]interface{}) error {
var err error
txManagerOnce.Do(func() {
maxVersions := 10
if config != nil {
if v, ok := config["max_versions"].(int); ok && v > 0 {
maxVersions = v
}
}
globalTxManager = &TransactionManager{
nextTxID: atomic.Uint64{},
walPath: walPath,
checkpointInterval: 300,
lastCheckpoint: time.Now().Unix(),
maxVersions: maxVersions,
visibilityMap: NewVisibilityMap(VisibilityMapSize),
readCache: NewReadTimestampCache(10000, 5*time.Minute),
distCoord: NewDistributedTransactionCoordinator(30 * time.Second),
deadlockDetector: NewDeadlockDetector(DeadlockCheckInterval, DefaultTxTimeout),
stats: &TransactionStats{StartTime: time.Now()},
mvccManager: NewMVCCManager(MaxVersionsPerDoc, VersionRetentionDays),
}
globalTxManager.nextTxID.Store(1)
var walErr error
// Используем fsync по умолчанию
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
}
// GetTransactionManager возвращает глобальный менеджер транзакций.
func GetTransactionManager() *TransactionManager {
return globalTxManager
}
// SetTransactionLogger - устанавливает логгер для транзакций.
func SetTransactionLogger(logger LoggerInterface) {
if globalTxManager != nil {
globalTxManager.logger = logger
if globalTxManager.wal != nil {
globalTxManager.wal.logger = logger
}
if globalTxManager.deadlockDetector != nil {
globalTxManager.deadlockDetector.SetLogger(logger)
}
}
}
// SetGlobalStorage - устанавливает глобальное хранилище.
func SetGlobalStorage(s *Storage) {
globalStorage = s
}
// GetGlobalStorage - возвращает глобальное хранилище.
func GetGlobalStorage() *Storage {
return globalStorage
}
// BeginTransaction - начинает новую транзакцию.
func BeginTransaction() *Transaction {
if globalTxManager == nil {
InitTransactionManager("futriis.wal")
}
return BeginTransactionWithOptions(&TransactionOptions{
Timeout: DefaultTxTimeout,
})
}
// BeginTransactionWithOptions - начинает транзакцию с опциями.
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)
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))
LogAudit("TIMEOUT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
"timeout_ms": options.Timeout.Milliseconds(),
})
}
})
}
LogAudit("START", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
"start_time": tx.StartTime,
"timeout_ms": options.Timeout.Milliseconds(),
"distributed": options.IsDistributed,
})
return tx
}
// BeginTransactionWithTimeout - начинает транзакцию с таймаутом.
func BeginTransactionWithTimeout(timeout time.Duration) *Transaction {
return BeginTransactionWithOptions(&TransactionOptions{
Timeout: timeout,
})
}
// BeginDistributedTransaction - начинает распределённую транзакцию.
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 - коммитит текущую транзакцию.
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")
}
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 {
// Создаём MVCC версию при коммите
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))
txRecord := &TransactionRecord{
ID: tx.ID,
State: TransactionCommitted,
Timestamp: time.Now().UnixMilli(),
Operations: tx.Operations,
}
data, err := json.Marshal(txRecord)
if err == nil {
walRecord := &WALRecord{
Type: 1,
Data: data,
}
globalTxManager.wal.Write(walRecord)
// Принудительная синхронизация WAL для ACID
globalTxManager.wal.Sync()
}
LogAudit("COMMIT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), 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
}
// AbortCurrentTransaction - отменяет текущую транзакцию.
func AbortCurrentTransaction() error {
txVal := currentTx.Load()
if txVal == nil {
return fmt.Errorf("no active transaction")
}
tx := txVal.(*Transaction)
tx.State.Store(int32(TransactionAborted))
LogAudit("ABORT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), 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
}
// CommitDistributedTransaction - коммитит распределённую транзакцию.
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")
}
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()
}
LogAudit("COMMIT_DISTRIBUTED", "TRANSACTION", fmt.Sprintf("%d", txID), map[string]interface{}{
"nodes": tx.Nodes,
})
return nil
}
// AbortDistributedTransaction - отменяет распределённую транзакцию.
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()
}
LogAudit("ABORT_DISTRIBUTED", "TRANSACTION", fmt.Sprintf("%d", txID), map[string]interface{}{
"nodes": tx.Nodes,
})
return nil
}
// applyOperation - применяет операцию к хранилищу.
func applyOperation(op Operation) error {
if globalStorage == nil {
return fmt.Errorf("storage not initialized")
}
db, err := globalStorage.GetDatabase(op.Database)
if err != nil {
return fmt.Errorf("database not found: %s", op.Database)
}
coll, err := db.GetCollection(op.Collection)
if err != nil {
return fmt.Errorf("collection not found: %s", op.Collection)
}
switch op.Type {
case "insert":
doc := NewDocumentWithID(op.DocumentID)
for k, v := range op.Data {
doc.SetField(k, v)
}
doc.Version = op.Version
// Создаём MVCC версию при вставке
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
}
// Создаём MVCC версию при обновлении
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)
}
return nil
}
// CreateSavepoint - создаёт точку сохранения в транзакции.
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()
LogAudit("SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
"savepoint": name,
"op_count": savepoint.OpCount,
})
return nil
}
// RollbackToSavepoint - откатывает транзакцию к точке сохранения.
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]
LogAudit("ROLLBACK_TO_SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
"savepoint": name,
})
return nil
}
// ReleaseSavepoint - освобождает точку сохранения.
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:]...)
LogAudit("RELEASE_SAVEPOINT", "TRANSACTION", fmt.Sprintf("%d", tx.ID), map[string]interface{}{
"savepoint": name,
})
return nil
}
}
return fmt.Errorf("savepoint '%s' not found", name)
}
// GetSavepoints - возвращает список всех savepoints.
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
}
// AddDocumentVersion - добавляет версию документа.
func (tm *TransactionManager) AddDocumentVersion(docID string, version *DocumentVersion) {
val, _ := tm.documentVersions.LoadOrStore(docID, make([]*DocumentVersion, 0))
versions := val.([]*DocumentVersion)
versions = append(versions, version)
if len(versions) > tm.maxVersions && tm.maxVersions > 0 {
versions = versions[len(versions)-tm.maxVersions:]
}
tm.documentVersions.Store(docID, versions)
if tm.visibilityMap != nil {
tm.visibilityMap.MarkVisible(docID, uint64(version.TxID), true)
}
}
// GetDocumentVersion - получает версию документа по временной метке.
func (tm *TransactionManager) GetDocumentVersion(docID string, timestamp int64) *Document {
if tm.readCache != nil {
if cached := tm.readCache.Get(docID, timestamp); cached != nil {
return cached
}
}
val, ok := tm.documentVersions.Load(docID)
if !ok {
return nil
}
versions := val.([]*DocumentVersion)
for i := len(versions) - 1; i >= 0; i-- {
if versions[i].Timestamp <= timestamp {
doc := versions[i].Document.Clone()
if tm.readCache != nil {
tm.readCache.Set(docID, timestamp, doc)
}
return doc
}
}
return nil
}
// checkpointLoop - периодическое создание чекпоинтов.
func (tm *TransactionManager) checkpointLoop() {
ticker := time.NewTicker(time.Duration(tm.checkpointInterval) * time.Second)
defer ticker.Stop()
for range ticker.C {
tm.createCheckpoint()
}
}
// createCheckpoint - создаёт чекпоинт.
func (tm *TransactionManager) createCheckpoint() {
if tm.wal == nil {
return
}
now := time.Now().Unix()
if now-tm.lastCheckpoint < tm.checkpointInterval {
return
}
checkpointPath := fmt.Sprintf("%s.checkpoint.%d", tm.walPath, now)
checkpoint := make(map[string]interface{})
checkpoint["timestamp"] = now
checkpoint["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
}
if err := os.WriteFile(checkpointPath, data, 0644); err != nil {
if tm.logger != nil {
tm.logger.Error(fmt.Sprintf("Failed to write checkpoint: %v", err))
}
return
}
tm.lastCheckpoint = now
if tm.logger != nil {
tm.logger.Info(fmt.Sprintf("Checkpoint created: %s", checkpointPath))
}
}
// versionCleanupLoop - периодическая очистка старых версий.
func (tm *TransactionManager) versionCleanupLoop() {
if tm.maxVersions <= 0 {
return
}
ticker := time.NewTicker(VersionPruneInterval)
defer ticker.Stop()
for range ticker.C {
cutoffTime := time.Now().AddDate(0, 0, -VersionRetentionDays).UnixMilli()
tm.documentVersions.Range(func(key, value interface{}) bool {
versions := value.([]*DocumentVersion)
if len(versions) <= tm.maxVersions {
return true
}
newVersions := make([]*DocumentVersion, 0, tm.maxVersions)
for _, v := range versions {
if v.Timestamp >= cutoffTime && len(newVersions) < tm.maxVersions {
newVersions = append(newVersions, v)
}
}
if len(newVersions) < len(versions) {
tm.documentVersions.Store(key, newVersions)
}
return true
})
}
}
// statsMonitor - мониторинг статистики.
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)
}
}
}
// startAsyncRecovery - запускает асинхронное восстановление.
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
}
}
}()
}
// IsRecoveryComplete - проверяет завершение восстановления.
func (tm *TransactionManager) IsRecoveryComplete() bool {
return tm.recoveryComplete.Load()
}
// GetRecoveryProgress - возвращает прогресс восстановления.
func (tm *TransactionManager) GetRecoveryProgress() map[string]interface{} {
if tm.recoveryManager == nil {
return map[string]interface{}{
"is_recovering": false,
"recovered": 0,
"complete": true,
}
}
stats := tm.recoveryManager.GetStats()
return map[string]interface{}{
"is_recovering": !tm.recoveryComplete.Load(),
"recovered": stats["recovered"],
"complete": tm.recoveryComplete.Load(),
"elapsed_ms": stats["elapsed_ms"],
}
}
// LockForBackup - блокирует транзакции для бэкапа.
func (tm *TransactionManager) LockForBackup() {
tm.backupLock.Lock()
tm.backupInProgress.Store(true)
}
// UnlockForBackup - разблокирует транзакции после бэкапа.
func (tm *TransactionManager) UnlockForBackup() {
tm.backupInProgress.Store(false)
tm.backupLock.Unlock()
}
// IsBackupInProgress - проверяет выполнение бэкапа.
func (tm *TransactionManager) IsBackupInProgress() bool {
return tm.backupInProgress.Load()
}
// GetTransactionStats - возвращает статистику транзакций.
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()
totalTimedOut := stats.TotalTimedOut.Load()
totalDeadlocks := stats.TotalDeadlocks.Load()
return map[string]interface{}{
"total_started": totalStarted,
"total_committed": totalCommitted,
"total_aborted": totalAborted,
"total_timed_out": totalTimedOut,
"total_deadlocks": totalDeadlocks,
"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(),
}
}
// StopTransactionManager - останавливает менеджер транзакций.
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
}
// HasActiveTransaction - проверяет наличие активной транзакции.
func HasActiveTransaction() bool {
return currentTx.Load() != nil
}
// GetCurrentTransactionID - возвращает ID текущей транзакции.
func GetCurrentTransactionID() string {
txVal := currentTx.Load()
if txVal == nil {
return ""
}
tx := txVal.(*Transaction)
return fmt.Sprintf("%d", tx.ID)
}
// GetActiveTransactions - возвращает список активных транзакций.
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
}
// GetTransactionByID - возвращает транзакцию по ID.
func GetTransactionByID(id string) (*Transaction, error) {
if globalTxManager == nil {
return nil, fmt.Errorf("transaction manager not initialized")
}
var txID TransactionID
fmt.Sscanf(id, "%d", &txID)
if val, ok := globalTxManager.activeTransactions.Load(txID); ok {
return val.(*Transaction), nil
}
return nil, fmt.Errorf("transaction not found")
}
// AddToTransaction - добавляет операцию в текущую транзакцию.
func AddToTransaction(coll *Collection, opType string, doc *Document) error {
txVal := currentTx.Load()
if txVal == nil {
return fmt.Errorf("no active transaction")
}
tx := txVal.(*Transaction)
if TransactionState(tx.State.Load()) != TransactionActive {
return fmt.Errorf("transaction is not active")
}
op := Operation{
Type: opType,
Database: coll.dbName,
Collection: coll.name,
DocumentID: doc.ID,
Data: doc.GetFields(),
Version: doc.Version,
}
tx.mu.Lock()
tx.Operations = append(tx.Operations, op)
tx.mu.Unlock()
return nil
}
// FindInTransaction - находит документ в контексте транзакции.
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)
}
// MVCCSnapshot - создаёт снапшот MVCC.
func MVCCSnapshot() uint64 {
return uint64(time.Now().UnixNano())
}
// CreateDocumentVersion - создаёт версию документа.
func CreateDocumentVersion(doc *Document, txID TransactionID) *DocumentVersion {
return &DocumentVersion{
Document: doc.Clone(),
Timestamp: time.Now().UnixMilli(),
TxID: txID,
}
}
// BeginTransactionOnCollection - начинает транзакцию на коллекции.
func BeginTransactionOnCollection(coll *Collection) error {
if globalTxManager == nil {
if err := InitTransactionManager("futriis.wal"); err != nil {
return err
}
}
tx := BeginTransaction()
if tx == nil {
return fmt.Errorf("failed to create transaction")
}
if globalTxManager.logger != nil {
globalTxManager.logger.Debug(fmt.Sprintf("Transaction %d started on collection %s.%s", tx.ID, coll.dbName, coll.name))
}
return nil
}
// CheckTransactionTimeout - проверяет таймаут транзакции.
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()
}
LogAudit("TIMEOUT_CHECK", "TRANSACTION", fmt.Sprintf("%d", txID), 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
}
Reference in New Issue Copy Permalink