fracmanager

package
v0.72.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 5, 2026 License: Apache-2.0 Imports: 41 Imported by: 0

Documentation

Index

Constants

View Source 
const StateFile = "storage-state.json"

Variables

View Source 
var (
	ErrFractionNotWritable = errors.New("fraction is not writable")
	ErrFractionSuspended   = errors.New("write operations temporarily suspended - database capacity exceeded")
)

Functions

func NewFracInfoCache added in v0.62.1 

func NewFracInfoCache(filePath string) *fracInfoCache

func NewFracInfoCacheFromDisk added in v0.62.1 

func NewFracInfoCacheFromDisk(filePath string) *fracInfoCache

func NewFractionRegistry added in v0.65.0 

func NewFractionRegistry(active *frac.Active, sealed []*frac.Sealed, remotes []*frac.Remote) (*fractionRegistry, error)

NewFractionRegistry creates and initializes a new fraction registry instance. Populates the registry with existing active, sealed and remote fractions. Rebuilds the complete fractions list in chronological order.

Types

type CacheMaintainer 

type CacheMaintainer struct {
	// contains filtered or unexported fields
}

func NewCacheMaintainer 

func NewCacheMaintainer(totalCacheSize, sortCacheSize uint64, metrics *CacheMaintainerMetrics) *CacheMaintainer

func (*CacheMaintainer) CreateDocBlockCache 

func (cm *CacheMaintainer) CreateDocBlockCache() *cache.Cache[[]byte]

func (*CacheMaintainer) CreateIndexCache 

func (cm *CacheMaintainer) CreateIndexCache() *frac.IndexCache

func (*CacheMaintainer) CreateSortDocsCache 

func (cm *CacheMaintainer) CreateSortDocsCache() *cache.Cache[[]byte]

func (*CacheMaintainer) Reset 

func (cm *CacheMaintainer) Reset()

Reset is used in tests only

func (*CacheMaintainer) RunCleanLoop 

func (cm *CacheMaintainer) RunCleanLoop(done <-chan struct{}, cleanupInterval, gcInterval time.Duration)

type CacheMaintainerMetrics 

type CacheMaintainerMetrics struct {
	HitsTotal       *prometheus.CounterVec
	MissTotal       *prometheus.CounterVec
	PanicsTotal     *prometheus.CounterVec
	LockWaitsTotal  *prometheus.CounterVec
	WaitsTotal      *prometheus.CounterVec
	ReattemptsTotal *prometheus.CounterVec
	SizeRead        *prometheus.CounterVec
	SizeOccupied    *prometheus.CounterVec
	SizeReleased    *prometheus.CounterVec
	EntriesReleased *prometheus.CounterVec
	MapsRecreated   *prometheus.CounterVec
	MissLatency     *prometheus.CounterVec

	Oldest            *prometheus.GaugeVec
	AddBuckets        *prometheus.CounterVec
	DelBuckets        *prometheus.CounterVec
	CleanGenerations  *prometheus.CounterVec
	ChangeGenerations *prometheus.CounterVec
}

func (*CacheMaintainerMetrics) GetCleanerMetrics 

func (m *CacheMaintainerMetrics) GetCleanerMetrics(cleanerLabel string) *cache.CleanerMetrics

func (*CacheMaintainerMetrics) GetLayerMetrics 

func (m *CacheMaintainerMetrics) GetLayerMetrics(layerName string) *cache.Metrics

type Config 

type Config struct {
	DataDir string

	FracSize  uint64
	TotalSize uint64
	CacheSize uint64

	SealingQueueLen uint64

	ReplayWorkers     int
	MaintenanceDelay  time.Duration
	CacheCleanupDelay time.Duration
	CacheGCDelay      time.Duration
	SealParams        common.SealParams
	SortCacheSize     uint64 // size for docs cache for active fraction
	Fraction          frac.Config
	MinSealFracSize   uint64

	OffloadingEnabled    bool
	OffloadingQueueSize  uint64
	OffloadingRetention  time.Duration
	OffloadingRetryDelay time.Duration
	// contains filtered or unexported fields
}

func FillConfigWithDefault 

func FillConfigWithDefault(config *Config) *Config

func (*Config) SuspendThreshold added in v0.68.0 

func (cfg *Config) SuspendThreshold() uint64

type Fetcher 

type Fetcher struct {
	// contains filtered or unexported fields
}

func NewFetcher 

func NewFetcher(maxWorkersNum int) *Fetcher

func (*Fetcher) FetchDocs 

func (f *Fetcher) FetchDocs(ctx context.Context, fracs List, ids []seq.IDSource, noSkipMasks bool) ([][]byte, error)

type FracManager 

type FracManager struct {
	// contains filtered or unexported fields
}

FracManager manages database fractions with lifecycle operations

func New added in v0.62.5 

func New(ctx context.Context, cfg *Config, s3cli *s3.Client, skipMaskProvider skipMaskProvider) (*FracManager, func(), error)

New creates and initializes a new fraction manager Starts all background workers:

  • indexer,
  • cache cleaner,
  • fraction rotation
  • stats updating

Returns the manager instance and a stop function to gracefully shutdown

func (*FracManager) AcquireFraction added in v0.71.0 

func (fm *FracManager) AcquireFraction(name string) (frac.Fraction, func(), bool)

func (*FracManager) Append 

func (fm *FracManager) Append(ctx context.Context, docs storage.DocBlock, metas storage.WalBlock) error

Append writes documents and metadata to the active fraction Implements retry logic in case of fraction sealing during write

func (*FracManager) Flags added in v0.62.3 

func (fm *FracManager) Flags() *StateManager

func (*FracManager) Fractions added in v0.62.5 

func (fm *FracManager) Fractions() List

func (*FracManager) Maintain added in v0.68.0 

func (fm *FracManager) Maintain(ctx context.Context, cfg *Config, wg *sync.WaitGroup)

Perform fraction maintenance (rotation, truncating, offloading, etc.)

func (*FracManager) Oldest added in v0.62.5 

func (fm *FracManager) Oldest() uint64

func (*FracManager) SealForcedForTests 

func (fm *FracManager) SealForcedForTests()

func (*FracManager) WaitIdleForTests added in v0.62.5 

func (fm *FracManager) WaitIdleForTests()

type List 

type List []frac.Fraction

func (List) FilterInRange 

func (l List) FilterInRange(from, to seq.MID) List

func (List) GetOldestFrac 

func (l List) GetOldestFrac() frac.Fraction

func (List) GetTotalSize 

func (l List) GetTotalSize() uint64

func (List) Names added in v0.71.0 

func (l List) Names() []string

func (*List) Shift 

func (l *List) Shift(n int) []frac.Fraction

func (List) Sort 

func (l List) Sort(order seq.DocsOrder)

type Loader added in v0.62.3 

type Loader struct {
	// contains filtered or unexported fields
}

Loader is responsible for loading and initializing fractions from filesystem Coordinates the process of discovering, validating, and loading all fraction types

func NewLoader 

func NewLoader(config *Config, provider *fractionProvider, infoCache *fracInfoCache) *Loader

NewLoader creates a new fraction loader Initialized at system startup to prepare data

func (*Loader) Load added in v0.62.3 

func (l *Loader) Load(ctx context.Context) (*fractionRegistry, error)

Load is the main method for loading all fractions Coordinates the entire process: discovery, validation, recovery, and ordering

type Searcher 

type Searcher struct {
	// contains filtered or unexported fields
}

func NewSearcher 

func NewSearcher(maxWorkersNum int, cfg SearcherCfg) *Searcher

func (*Searcher) SearchDocs 

func (s *Searcher) SearchDocs(ctx context.Context, fracs []frac.Fraction, params processor.SearchParams, tr *querytracer.Tracer) (*seq.QPR, error)

type SearcherCfg 

type SearcherCfg struct {
	MaxFractionHits       int // the maximum number of fractions used in the search
	FractionsPerIteration int
	SlowLogThreshold      time.Duration
	MaxQprMemory          int // max heap memory a single QPR can use. 0 if no limit set
}

type StateManager added in v0.62.3 

type StateManager struct {
	// contains filtered or unexported fields
}

StateManager manages storage state with thread safety

func NewStateManager added in v0.62.3 

func NewStateManager(dataDir string, defaultState StorageState) (*StateManager, error)

NewStateManager creates a new storage state manager

func (*StateManager) IsCapacityExceeded added in v0.62.3 

func (m *StateManager) IsCapacityExceeded() bool

IsCapacityExceeded returns storage capacity exceeded flag

type StorageState added in v0.62.3 

type StorageState struct {
	CapacityExceeded bool `json:"capacity_exceeded"` // Storage capacity exceeded flag
}

type TaskManager added in v0.68.0 

type TaskManager struct {
	// contains filtered or unexported fields
}

TaskManager manages a collection of running background tasks Provides safe concurrent access to task tracking and cancellation.

func NewTaskManager added in v0.68.0 

func NewTaskManager() *TaskManager

func (*TaskManager) Cancel added in v0.68.0 

func (t *TaskManager) Cancel(id string)

Cancel cancels and waits for completion of a task by ID Returns immediately if task with given ID doesn't exist.

func (*TaskManager) Run added in v0.68.0 

func (t *TaskManager) Run(id string, ctx context.Context, action func(ctx context.Context)) (*task, error)

Run starts a new background task with the given ID and context. The task will be automatically removed when completed.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.