futriix/internal/storage/runtime_limits.go

/*
 * 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/runtime_limits.go
// Назначение: Ограничения на размер коллекции/документа в рантайме
//  Добавлен механизм eviction при нехватке памяти (OOM protection)
//  Eviction проверяет активные batch-операции перед удалением
//
// ИСПРАВЛЕНО (2026-10): после замены ACID-транзакций на batch-операции
// функция isDocInActiveTransaction ссылалась на удалённые globalTxManager
// и Transaction. Заменена на isDocInActiveBatch, который проверяет
// реестр активных batch'ей в globalBatchManager.
//
// Логика та же: не вытеснять документы, которые в данный момент
// участвуют в незакоммиченной batch-операции. Batch живёт коротко
// (создаётся, коммитится и удаляется в рамках одной функции), но
// между AddUpdate и Commit документ ещё не изменён, и eviction
// мог бы его удалить — защита сохраняется.

package storage

import (
	"fmt"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

// LoggerInterface определяет интерфейс для логирования
type LoggerInterface interface {
	Debug(msg string)
	Info(msg string)
	Error(msg string)
	Warn(msg string)
}

// EvictionPolicy определяет политику вытеснения
type EvictionPolicy int

const (
	EvictionNone EvictionPolicy = iota
	EvictionLRU
	EvictionTTL
	EvictionOldest
)

// RuntimeLimitsManager управляет runtime-ограничениями
type RuntimeLimitsManager struct {
	mu                   sync.RWMutex
	globalMaxDocSize     int64
	globalMaxCollSize    int64
	globalMaxDocsPerColl int64
	globalMaxMemory      int64
	collectionOverrides  map[string]*CollectionLimits
	metrics              *LimitMetrics
	logger               LoggerInterface
	enabled              bool
	evictionPolicy       EvictionPolicy
	memoryThreshold      float64
	evictionChan         chan string
	stopChan             chan struct{}
	wg                   sync.WaitGroup
}

// CollectionLimits содержит лимиты для конкретной коллекции
type CollectionLimits struct {
	MaxDocSize        int64
	MaxCollectionSize int64
	MaxDocuments      int64
	EvictionPolicy    EvictionPolicy
	LastUpdated       int64
}

// LimitMetrics хранит метрики ограничений
type LimitMetrics struct {
	RejectedBySize     atomic.Uint64
	RejectedByDocCount atomic.Uint64
	RejectedByCollSize atomic.Uint64
	RejectedByMemory   atomic.Uint64
	EvictedDocuments   atomic.Uint64
	EvictedBytes       atomic.Uint64
	LastCheckTime      atomic.Int64
	LastEvictionTime   atomic.Int64
	SkippedEviction    atomic.Uint64 // Пропущено из-за активных batch-операций
}

// RuntimeLimitsConfig содержит конфигурацию ограничений
type RuntimeLimitsConfig struct {
	Enabled              bool           `json:"enabled"`
	GlobalMaxDocSizeMB   int            `json:"global_max_doc_size_mb"`
	GlobalMaxCollSizeMB  int64          `json:"global_max_coll_size_mb"`
	GlobalMaxDocsPerColl int64          `json:"global_max_docs_per_coll"`
	GlobalMaxMemoryMB    int64          `json:"global_max_memory_mb"`
	EvictionPolicy       EvictionPolicy `json:"eviction_policy"`
	MemoryThreshold      float64        `json:"memory_threshold"`
}

// DefaultRuntimeLimitsConfig возвращает конфигурацию по умолчанию
func DefaultRuntimeLimitsConfig() *RuntimeLimitsConfig {
	return &RuntimeLimitsConfig{
		Enabled:              true,
		GlobalMaxDocSizeMB:   16,
		GlobalMaxCollSizeMB:  10240,
		GlobalMaxDocsPerColl: 10000000,
		GlobalMaxMemoryMB:    0,
		EvictionPolicy:       EvictionLRU,
		MemoryThreshold:      0.85,
	}
}

// NewRuntimeLimitsManager создаёт новый менеджер ограничений
func NewRuntimeLimitsManager(cfg *RuntimeLimitsConfig, logger LoggerInterface) *RuntimeLimitsManager {
	if cfg == nil {
		cfg = DefaultRuntimeLimitsConfig()
	}

	maxMemory := cfg.GlobalMaxMemoryMB * 1024 * 1024
	if maxMemory <= 0 {
		var memStats runtime.MemStats
		runtime.ReadMemStats(&memStats)
		maxMemory = int64(float64(memStats.Sys) * 0.8)
	}

	rlm := &RuntimeLimitsManager{
		globalMaxDocSize:     int64(cfg.GlobalMaxDocSizeMB) * 1024 * 1024,
		globalMaxCollSize:    cfg.GlobalMaxCollSizeMB * 1024 * 1024,
		globalMaxDocsPerColl: cfg.GlobalMaxDocsPerColl,
		globalMaxMemory:      maxMemory,
		collectionOverrides:  make(map[string]*CollectionLimits),
		metrics:              &LimitMetrics{},
		logger:               logger,
		enabled:              cfg.Enabled,
		evictionPolicy:       cfg.EvictionPolicy,
		memoryThreshold:      cfg.MemoryThreshold,
		evictionChan:         make(chan string, 100),
		stopChan:             make(chan struct{}),
	}

	rlm.wg.Add(1)
	go rlm.memoryMonitorLoop()

	if logger != nil {
		logger.Debug(fmt.Sprintf("Runtime limits manager initialized: maxDoc=%dMB, maxColl=%dMB, maxDocs=%d, maxMemory=%dMB, eviction=%d",
			cfg.GlobalMaxDocSizeMB, cfg.GlobalMaxCollSizeMB, cfg.GlobalMaxDocsPerColl, maxMemory/(1024*1024), cfg.EvictionPolicy))
	}

	return rlm
}

func (rlm *RuntimeLimitsManager) memoryMonitorLoop() {
	defer rlm.wg.Done()
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			if !rlm.enabled {
				continue
			}
			var memStats runtime.MemStats
			runtime.ReadMemStats(&memStats)
			currentUsage := int64(memStats.Alloc)
			usageRatio := float64(currentUsage) / float64(rlm.globalMaxMemory)

			if usageRatio > rlm.memoryThreshold {
				if rlm.logger != nil {
					rlm.logger.Warn(fmt.Sprintf("Memory usage %.2f%% exceeds threshold %.2f%%, triggering eviction",
						usageRatio*100, rlm.memoryThreshold*100))
				}
				rlm.triggerEviction()
			}
		case <-rlm.stopChan:
			return
		}
	}
}

func (rlm *RuntimeLimitsManager) triggerEviction() {
	select {
	case rlm.evictionChan <- "global":
	default:
	}
}

// isDocInActiveBatch проверяет, участвует ли документ в активной
// batch-операции.
//
// ИСПРАВЛЕНО (2026-10): ранее метод назывался isDocInActiveTransaction
// и обращался к globalTxManager и Transaction, которые были удалены при
// переходе на batch-операции. Теперь проверяем реестр активных batch'ей
// в globalBatchManager (см. transactions.go).
//
// Защита сохраняется: batch живёт коротко (создаётся и коммитится в
// рамках одной функции), но между AddUpdate и Commit документ ещё не
// изменён, и eviction мог бы его удалить.
func (rlm *RuntimeLimitsManager) isDocInActiveBatch(docID, dbName, collName string) bool {
	bm := GetBatchManager()
	if bm == nil {
		return false
	}
	return bm.IsDocInActiveBatch(docID, dbName, collName)
}

// EvictFromCollection выполняет вытеснение документов из коллекции.
//
// Проверка активных batch-операций перед удалением.
func (rlm *RuntimeLimitsManager) EvictFromCollection(coll *Collection, targetBytes int64) (int64, error) {
	if !rlm.enabled {
		return 0, nil
	}

	evictedBytes := int64(0)
	evictedCount := int64(0)
	skippedCount := int64(0)

	docs := coll.GetAllDocumentsIncludingDeleted()

	// Сначала вытесняем удалённые документы
	for _, doc := range docs {
		if evictedBytes >= targetBytes {
			break
		}

		if doc.IsDeleted() {
			// Проверяем активные batch-операции
			if rlm.isDocInActiveBatch(doc.ID, coll.DBName(), coll.Name()) {
				skippedCount++
				continue
			}

			size := doc.OriginalSize
			if size == 0 {
				size = 1024
			}

			if err := coll.PermanentDelete(doc.ID); err == nil {
				evictedBytes += size
				evictedCount++
			}
		}
	}

	// Затем вытесняем самые старые
	if evictedBytes < targetBytes && rlm.evictionPolicy == EvictionLRU {
		for _, doc := range docs {
			if evictedBytes >= targetBytes {
				break
			}

			if !doc.IsDeleted() {
				// Проверяем активные batch-операции
				if rlm.isDocInActiveBatch(doc.ID, coll.DBName(), coll.Name()) {
					skippedCount++
					continue
				}

				size := doc.OriginalSize
				if size == 0 {
					size = 1024
				}

				if coll.metadata.Settings.SoftDelete {
					if err := coll.Delete(doc.ID); err == nil {
						evictedBytes += size
						evictedCount++
					}
				} else {
					if err := coll.PermanentDelete(doc.ID); err == nil {
						evictedBytes += size
						evictedCount++
					}
				}
			}
		}
	}

	rlm.metrics.EvictedDocuments.Add(uint64(evictedCount))
	rlm.metrics.EvictedBytes.Add(uint64(evictedBytes))
	if skippedCount > 0 {
		rlm.metrics.SkippedEviction.Add(uint64(skippedCount))
	}
	rlm.metrics.LastEvictionTime.Store(time.Now().UnixMilli())

	if rlm.logger != nil {
		rlm.logger.Info(fmt.Sprintf("Evicted %d documents (%d bytes) from collection %s.%s (skipped %d in active batches)",
			evictedCount, evictedBytes, coll.DBName(), coll.Name(), skippedCount))
	}

	return evictedBytes, nil
}

func (rlm *RuntimeLimitsManager) ValidateDocumentSize(dbName, collName string, docSize int64) error {
	if !rlm.enabled {
		return nil
	}

	limit := rlm.getCollectionLimit(dbName, collName)
	maxSize := rlm.globalMaxDocSize
	if limit != nil && limit.MaxDocSize > 0 {
		maxSize = limit.MaxDocSize
	}

	if docSize > maxSize {
		rlm.metrics.RejectedBySize.Add(1)
		return fmt.Errorf("document size %d bytes exceeds limit %d bytes", docSize, maxSize)
	}
	return nil
}

func (rlm *RuntimeLimitsManager) ValidateCollectionSize(coll *Collection, newDocSize int64) error {
	if !rlm.enabled {
		return nil
	}

	limit := rlm.getCollectionLimit(coll.DBName(), coll.Name())
	maxSize := rlm.globalMaxCollSize
	if limit != nil && limit.MaxCollectionSize > 0 {
		maxSize = limit.MaxCollectionSize
	}

	currentSize := coll.Size()
	if currentSize+newDocSize > maxSize {
		neededBytes := currentSize + newDocSize - maxSize
		if evicted, err := rlm.EvictFromCollection(coll, neededBytes); err == nil && evicted >= neededBytes {
			return nil
		}

		rlm.metrics.RejectedByCollSize.Add(1)
		return fmt.Errorf("collection size would exceed limit %d bytes (current: %d, new: %d)",
			maxSize, currentSize, newDocSize)
	}
	return nil
}

func (rlm *RuntimeLimitsManager) ValidateDocumentCount(coll *Collection) error {
	if !rlm.enabled {
		return nil
	}

	limit := rlm.getCollectionLimit(coll.DBName(), coll.Name())
	maxDocs := rlm.globalMaxDocsPerColl
	if limit != nil && limit.MaxDocuments > 0 {
		maxDocs = limit.MaxDocuments
	}

	currentCount := coll.Count()
	if currentCount >= maxDocs {
		targetBytes := int64((currentCount - maxDocs + 1) * 1024)
		if evicted, err := rlm.EvictFromCollection(coll, targetBytes); err == nil && evicted > 0 {
			if coll.Count() < maxDocs {
				return nil
			}
		}

		rlm.metrics.RejectedByDocCount.Add(1)
		return fmt.Errorf("collection has reached maximum document count %d", maxDocs)
	}
	return nil
}

func (rlm *RuntimeLimitsManager) CheckMemoryUsage() error {
	if !rlm.enabled {
		return nil
	}

	var memStats runtime.MemStats
	runtime.ReadMemStats(&memStats)

	currentUsage := int64(memStats.Alloc)
	if currentUsage > rlm.globalMaxMemory {
		rlm.metrics.RejectedByMemory.Add(1)
		return fmt.Errorf("memory usage %d bytes exceeds limit %d bytes", currentUsage, rlm.globalMaxMemory)
	}
	return nil
}

func (rlm *RuntimeLimitsManager) getCollectionLimit(dbName, collName string) *CollectionLimits {
	rlm.mu.RLock()
	defer rlm.mu.RUnlock()

	key := fmt.Sprintf("%s.%s", dbName, collName)
	if limits, ok := rlm.collectionOverrides[key]; ok {
		return limits
	}
	return nil
}

func (rlm *RuntimeLimitsManager) SetCollectionLimits(dbName, collName string, maxDocSizeMB int, maxCollSizeMB int64, maxDocuments int64) {
	rlm.mu.Lock()
	defer rlm.mu.Unlock()

	key := fmt.Sprintf("%s.%s", dbName, collName)
	rlm.collectionOverrides[key] = &CollectionLimits{
		MaxDocSize:        int64(maxDocSizeMB) * 1024 * 1024,
		MaxCollectionSize: maxCollSizeMB * 1024 * 1024,
		MaxDocuments:      maxDocuments,
		EvictionPolicy:    rlm.evictionPolicy,
		LastUpdated:       time.Now().UnixMilli(),
	}

	if rlm.logger != nil {
		rlm.logger.Info(fmt.Sprintf("Set limits for %s: maxDoc=%dMB, maxColl=%dMB, maxDocs=%d",
			key, maxDocSizeMB, maxCollSizeMB, maxDocuments))
	}
}

func (rlm *RuntimeLimitsManager) RemoveCollectionLimits(dbName, collName string) {
	rlm.mu.Lock()
	defer rlm.mu.Unlock()

	key := fmt.Sprintf("%s.%s", dbName, collName)
	delete(rlm.collectionOverrides, key)

	if rlm.logger != nil {
		rlm.logger.Info(fmt.Sprintf("Removed limits override for %s", key))
	}
}

func (rlm *RuntimeLimitsManager) GetMetrics() map[string]interface{} {
	return map[string]interface{}{
		"rejected_by_size":         rlm.metrics.RejectedBySize.Load(),
		"rejected_by_doc_count":    rlm.metrics.RejectedByDocCount.Load(),
		"rejected_by_coll_size":    rlm.metrics.RejectedByCollSize.Load(),
		"rejected_by_memory":       rlm.metrics.RejectedByMemory.Load(),
		"evicted_documents":        rlm.metrics.EvictedDocuments.Load(),
		"evicted_bytes":            rlm.metrics.EvictedBytes.Load(),
		"skipped_eviction":         rlm.metrics.SkippedEviction.Load(),
		"last_eviction_time":       rlm.metrics.LastEvictionTime.Load(),
		"global_max_doc_size_mb":   rlm.globalMaxDocSize / (1024 * 1024),
		"global_max_coll_size_mb":  rlm.globalMaxCollSize / (1024 * 1024),
		"global_max_docs_per_coll": rlm.globalMaxDocsPerColl,
		"global_max_memory_mb":     rlm.globalMaxMemory / (1024 * 1024),
		"eviction_policy":          rlm.evictionPolicy,
		"memory_threshold":         rlm.memoryThreshold,
		"enabled":                  rlm.enabled,
	}
}

func (rlm *RuntimeLimitsManager) Stop() {
	close(rlm.stopChan)
	rlm.wg.Wait()
}