feat(fracmanager): add handling for upload queue overflow and disk space exhaustion (#266)

Author: eguguchkin

cmd/seq-db/seq-db.go

@@ -253,6 +253,7 @@ func startStore(
 			DataDir:           cfg.Storage.DataDir,
 			FracSize:          uint64(cfg.Storage.FracSize),
 			TotalSize:         uint64(cfg.Storage.TotalSize),
+			SealingQueueLen:   uint64(cfg.Storage.SealingQueueLen),
 			CacheSize:         uint64(cfg.Resources.CacheSize),
 			SortCacheSize:     uint64(cfg.Resources.SortDocsCacheSize),
 			ReplayWorkers:     cfg.Resources.ReplayWorkers,
@@ -281,8 +282,10 @@ func startStore(
 				SkipSortDocs: !cfg.DocsSorting.Enabled,
 				KeepMetaFile: false,
 			},
-			OffloadingEnabled:   cfg.Offloading.Enabled,
-			OffloadingRetention: cfg.Offloading.Retention,
+			OffloadingEnabled:    cfg.Offloading.Enabled,
+			OffloadingRetention:  cfg.Offloading.Retention,
+			OffloadingRetryDelay: cfg.Offloading.RetryDelay,
+			OffloadingQueueSize:  uint64(float64(cfg.Storage.TotalSize) * cfg.Offloading.QueueSizePercent / 100),
 		},
 		API: storeapi.APIConfig{
 			StoreMode: configMode,
@@ -342,7 +345,6 @@ func initS3Client(cfg config.Config) *s3.Client {
 		cfg.Offloading.SecretKey,
 		cfg.Offloading.Region,
 		cfg.Offloading.Bucket,
-		cfg.Offloading.RetryCount,
 	)
 
 	if err != nil {

config/config.go

@@ -63,6 +63,11 @@ type Config struct {
 		// TotalSize specifies upper bound of how much disk space can be occupied
 		// by sealed fractions before they get deleted (or offloaded).
 		TotalSize Bytes `config:"total_size" default:"1GiB"`
+		// SealingQueueLen defines the maximum length of the sealing queue.
+		// If the queue size exceeds this limit, writing to the store will be paused,
+		// and bulk requests will start returning errors.
+		// A value of zero disables this limit, allowing writes to proceed unconditionally.
+		SealingQueueLen int `config:"sealing_queue_len" default:"10"`
 	} `config:"storage"`
 
 	Cluster struct {
@@ -241,9 +246,12 @@ type Config struct {
 		// SecretKey configures S3 Secret Key for S3 client.
 		// You can learn more about secret keys [here](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html).
 		SecretKey string `config:"secret_key"`
-		// RetryCount sets [RetryMaxAttempts] for S3 client which is applied for all API calls.
-		// Be aware that fraction is suicided when offloading attempts exceeds [RetryCount].
-		RetryCount int `config:"retry_count" default:"5"`
+		// Specifies the percentage of total local dataset size allocated to the offloading queue.
+		// Note: When the queue overflows, the oldest fraction of data is automatically removed.
+		// This automatic removal is disabled when set to zero.
+		QueueSizePercent float64 `config:"queue_size_percent" default:"5"`
+		// Delay duration between consecutive offloading retries
+		RetryDelay time.Duration `config:"retry_delay" default:"2s"`
 	} `config:"offloading"`
 
 	AsyncSearch struct {

config/validation.go

@@ -64,9 +64,11 @@ func (c *Config) storeValidations() []validateFn {
 		greaterThan("resources.search_workers", 0, c.Resources.SearchWorkers),
 		greaterThan("resources.replay_workers", 0, c.Resources.ReplayWorkers),
 		greaterThan("resources.cache_size", 0, c.Resources.CacheSize),
+		greaterThan("storage.sealing_queue_len", -1, c.Storage.SealingQueueLen),
 
 		inRange("compression.sealed_zstd_compression_level", -7, 22, c.Compression.SealedZstdCompressionLevel),
 		inRange("compression.doc_block_zstd_compression_level", -7, 22, c.Compression.DocBlockZstdCompressionLevel),
+		inRange("offloading.queue_size_percent", 0, 100, c.Offloading.QueueSizePercent),
 
 		greaterThan("experimental.max_regex_tokens_check", -1, c.Experimental.MaxRegexTokensCheck),
 	}

config/validation_test.go

@@ -0,0 +1,119 @@
+package config
+
+import (
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidation(t *testing.T) {
+	base := `storage:
+  data_dir: /seq-db-data
+  frac_size: 16MiB
+  total_size: 10GiB
+
+mapping:
+  path: /configs/mapping.yaml
+
+resources:
+  cache_size: 2GiB
+
+limits:
+  query_rate: 1024
+  search_requests: 1024
+  bulk_requests: 128
+  inflight_bulks: 128
+  doc_size: 1MiB
+`
+
+	baseCfg := createCfgFile(t, base)
+
+	tests := []struct {
+		name      string
+		cfg       string
+		env       map[string]string
+		expectErr bool
+	}{
+		{
+			name:      "Invalid storage.sealing_queue_len 1",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_STORAGE_SEALING_QUEUE_LEN": "-1"},
+			expectErr: true,
+		},
+		{
+			name:      "Valid storage.sealing_queue_len 2",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_STORAGE_SEALING_QUEUE_LEN": "0"},
+			expectErr: false,
+		},
+		{
+			name:      "Valid storage.sealing_queue_len 3",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_STORAGE_SEALING_QUEUE_LEN": "100"},
+			expectErr: false,
+		},
+
+		{
+			name:      "Invalid offloading.queue_size_percent 1",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_OFFLOADING_QUEUE_SIZE_PERCENT": "-1"},
+			expectErr: true,
+		},
+		{
+			name:      "Invalid offloading.queue_size_percent 2",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_OFFLOADING_QUEUE_SIZE_PERCENT": "100.1"},
+			expectErr: true,
+		},
+		{
+			name:      "Valid offloading.queue_size_percent 3",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_OFFLOADING_QUEUE_SIZE_PERCENT": "0"},
+			expectErr: false,
+		},
+		{
+			name:      "Valid offloading.queue_size_percent 4",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_OFFLOADING_QUEUE_SIZE_PERCENT": "100"},
+			expectErr: false,
+		},
+		{
+			name:      "Valid offloading.queue_size_percent 5",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_OFFLOADING_QUEUE_SIZE_PERCENT": "50"},
+			expectErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for k, v := range tt.env {
+				t.Setenv(k, v)
+			}
+
+			c, err := Parse(tt.cfg)
+			assert.NoError(t, err)
+
+			res := c.Validate("store")
+			if tt.expectErr {
+				assert.Error(t, res)
+			} else {
+				assert.NoError(t, res)
+			}
+		})
+	}
+
+}
+
+func createCfgFile(t *testing.T, data string) string {
+	f := path.Join(t.TempDir(), "config.yaml")
+	err := os.WriteFile(f, []byte(data), 0o666)
+	assert.NoError(t, err)
+
+	abs, err := filepath.Abs(f)
+	assert.NoError(t, err)
+	return abs
+}

frac/fraction_test.go

@@ -2134,7 +2134,6 @@ func (s *RemoteFractionTestSuite) SetupTest() {
 			"SECRET_KEY",
 			"eu-west-3",
 			bucketName,
-			3,
 		)
 		s.Require().NoError(err, "s3 client setup failed")
 

fracmanager/config.go

@@ -19,6 +19,9 @@ type Config struct {
 	TotalSize uint64
 	CacheSize uint64
 
+	suspendThreshold uint64
+	SealingQueueLen  uint64
+
 	ReplayWorkers     int
 	MaintenanceDelay  time.Duration
 	CacheCleanupDelay time.Duration
@@ -28,8 +31,10 @@ type Config struct {
 	Fraction          frac.Config
 	MinSealFracSize   uint64
 
-	OffloadingEnabled   bool
-	OffloadingRetention time.Duration
+	OffloadingEnabled    bool
+	OffloadingQueueSize  uint64
+	OffloadingRetention  time.Duration
+	OffloadingRetryDelay time.Duration
 }
 
 func FillConfigWithDefault(config *Config) *Config {
@@ -82,3 +87,14 @@ func FillConfigWithDefault(config *Config) *Config {
 
 	return config
 }
+
+func (cfg *Config) SuspendThreshold() uint64 {
+	if cfg.suspendThreshold == 0 {
+		cfg.suspendThreshold = cfg.TotalSize
+		cfg.suspendThreshold += cfg.TotalSize / 100 // small buffer
+		if cfg.OffloadingEnabled {
+			cfg.suspendThreshold += cfg.OffloadingQueueSize // offloading queue size
+		}
+	}
+	return cfg.suspendThreshold
+}

fracmanager/fracmanager.go

@@ -128,6 +128,18 @@ func (fm *FracManager) Append(ctx context.Context, docs, metas storage.DocBlock)
 	}
 }
 
+// Perform fraction maintenance (rotation, truncating, offloading, etc.)
+func (fm *FracManager) Maintain(ctx context.Context, cfg *Config, wg *sync.WaitGroup) {
+	n := time.Now()
+	logger.Debug("maintenance iteration started")
+
+	fm.mu.Lock()
+	defer fm.mu.Unlock()
+
+	fm.lc.Maintain(ctx, cfg, wg)
+	logger.Debug("maintenance iteration finished", zap.Int64("took_ms", time.Since(n).Milliseconds()))
+}
+
 // startCacheWorker starts background cache garbage collection
 func startCacheWorker(ctx context.Context, cfg *Config, cache *CacheMaintainer, wg *sync.WaitGroup) {
 	wg.Add(1)
@@ -161,21 +173,18 @@ func startStatsWorker(ctx context.Context, reg *fractionRegistry, wg *sync.WaitG
 // startMaintWorker starts periodic fraction maintenance operations
 func startMaintWorker(ctx context.Context, cfg *Config, fm *FracManager, wg *sync.WaitGroup) {
 	wg.Add(1)
+	maintWg := sync.WaitGroup{}
+
 	go func() {
 		defer wg.Done()
 
 		logger.Info("maintenance loop is started")
 		// Run maintenance at configured interval
 		util.RunEvery(ctx.Done(), cfg.MaintenanceDelay, func() {
-			n := time.Now()
-			logger.Debug("maintenance iteration started")
-			fm.mu.Lock()
-			// Perform fraction maintenance (rotation, truncating, offloading, etc.)
-			fm.lc.Maintain(ctx, cfg, wg)
-			fm.mu.Unlock()
-			logger.Debug("maintenance iteration finished", zap.Int64("took_ms", time.Since(n).Milliseconds()))
+			fm.Maintain(ctx, cfg, &maintWg)
 		})
-		logger.Info("waiting maintenance complete background tasks")
+		logger.Info("waiting maintenance complete background tasks...")
+		maintWg.Wait()
 		logger.Info("maintenance loop is stopped")
 	}()
 }

fracmanager/fraction_provider_test.go

@@ -26,7 +26,7 @@ func setupS3Client(t testing.TB) (*s3.Client, func()) {
 	err := s3Backend.CreateBucket(bucketName)
 	require.NoError(t, err, "create bucket failed")
 
-	s3cli, err := s3.NewClient(s3server.URL, "ACCESS_KEY", "SECRET_KEY", "eu-west-3", bucketName, 3)
+	s3cli, err := s3.NewClient(s3server.URL, "ACCESS_KEY", "SECRET_KEY", "eu-west-3", bucketName)
 	require.NoError(t, err, "s3 client setup failed")
 
 	return s3cli, s3server.Close