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

Author: eguguchkin

cmd/seq-db/seq-db.go

@@ -280,8 +280,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,

config/config.go

@@ -234,8 +234,13 @@ type Config struct {
 		// 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 {

fracmanager/config.go

@@ -28,8 +28,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 {

fracmanager/fracmanager.go

@@ -81,7 +81,7 @@ func New(ctx context.Context, cfg *Config, s3cli *s3.Client) (*FracManager, Stop
 
 		// Freeze active fraction to prevent new writes
 		active := lc.registry.Active()
-		if err := active.Freeze(); err != nil {
+		if err := active.Finalize(); err != nil {
 			logger.Fatal("shutdown fraction freezing error", zap.Error(err))
 		}
 
@@ -100,11 +100,6 @@ func New(ctx context.Context, cfg *Config, s3cli *s3.Client) (*FracManager, Stop
 	return &fm, stop, nil
 }
 
-// Writer returns the active fraction for write operations
-func (fm *FracManager) Writer() *activeProxy {
-	return fm.lc.registry.Active()
-}
-
 func (fm *FracManager) Fractions() List {
 	return fm.lc.registry.AllFractions()
 }
@@ -119,24 +114,27 @@ func (fm *FracManager) Flags() *StateManager {
 
 // Active returns the currently active fraction
 func (fm *FracManager) Active() frac.Fraction {
-	return fm.Writer().proxy
+	return fm.lc.registry.Active().proxy
 }
 
 // Append writes documents and metadata to the active fraction
 // Implements retry logic in case of fraction sealing during write
 func (fm *FracManager) Append(ctx context.Context, docs, metas storage.DocBlock) error {
-	var err error
 	for {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		default:
-			// Attempt to append to active fraction
-			if err = fm.Writer().Append(docs, metas); err == nil {
-				return nil
+			// Try to append data to the currently active fraction
+			err := fm.lc.registry.Active().Append(docs, metas)
+			if err != nil {
+				logger.Info("append fail", zap.Error(err))
+				if err == ErrFractionNotWritable {
+					// Fraction is currently being sealed, retry the operation
+					continue
+				}
 			}
-			// Log error (typically occurs when fraction is sealed during write)
-			logger.Info("append fail", zap.Error(err))
+			return err
 		}
 	}
 }

fracmanager/fracmanager_for_tests.go

@@ -3,13 +3,13 @@ package fracmanager
 import "sync"
 
 func (fm *FracManager) WaitIdleForTests() {
-	fm.Writer().WaitWriteIdle()
+	fm.lc.registry.Active().WaitWriteIdle()
 }
 
 func (fm *FracManager) SealForcedForTests() {
 	wg := sync.WaitGroup{}
 	fm.mu.Lock() // todo: get rid of mutex after removing SealForcedForTests method
-	fm.lc.Rotate(0, &wg)
+	fm.lc.ManageRotation(0, 0, &wg)
 	fm.mu.Unlock()
 
 	wg.Wait()

fracmanager/fraction_registry.go

@@ -110,29 +110,47 @@ func (r *fractionRegistry) OldestLocal() uint64 {
 	return r.oldestLocal
 }
 
-// Rotate completes the current active fraction and starts a new one.
+// RotateIfNeeded completes the current active fraction and starts a new one.
 // Moves previous active fraction to sealing queue.
 // Updates statistics and maintains chronological order.
 // Should be called when creating a new fraction.
-func (r *fractionRegistry) Rotate(newActive *activeProxy) (*activeProxy, error) {
+func (r *fractionRegistry) RotateIfNeeded(activeLimit uint64, activeProvider func() *activeProxy) (*activeProxy, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
-	prevActive := r.active
-	r.sealing = append(r.sealing, prevActive)
-	r.addNewActive(newActive)
+	if r.active.instance.Info().DocsOnDisk < activeLimit {
+		return nil, nil
+	}
+
+	oldActive := r.active
+	r.sealing = append(r.sealing, oldActive)
+	r.addNewActive(activeProvider())
 
 	// Freezing rotated fraction
 	// todo can be too long!
-	if err := prevActive.Freeze(); err != nil {
-		return prevActive, err
+	if err := oldActive.Finalize(); err != nil {
+		return oldActive, err
 	}
 
 	// Update statistics
-	finalInfo := prevActive.instance.Info()
+	finalInfo := oldActive.instance.Info()
 	r.stats.sealing.Add(finalInfo)
 
-	return prevActive, nil
+	return oldActive, nil
+}
+
+func (r *fractionRegistry) SuspendIfOverCapacity(maxTotalSize uint64) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	currentTotalUsage := r.active.instance.Info().FullSize() +
+		r.stats.locals.totalSizeOnDisk +
+		r.stats.sealing.totalSizeOnDisk +
+		r.stats.offloading.totalSizeOnDisk
+
+	val := currentTotalUsage > maxTotalSize
+	r.active.Suspend(val)
+	return val
 }
 
 // addNewActive sets a new active fraction and updates the complete fractions list.
@@ -232,6 +250,38 @@ func (r *fractionRegistry) EvictRemote(retention time.Duration) []*remoteProxy {
 	return evicted
 }
 
+// EvictOverflowed removes oldest fractions from offloading queue when it exceeds size limit.
+// Selects fractions that haven't finished offloading yet to minimize data loss.
+// Used when offloading queue grows too large due to slow remote storage performance.
+func (r *fractionRegistry) EvictOverflowed(sizeLimit uint64) []*sealedProxy {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	// Fast path: skip processing if within size limits
+	if r.stats.offloading.totalSizeOnDisk <= sizeLimit {
+		return nil
+	}
+
+	i := 0
+	evicted := []*sealedProxy{}
+	// filter fractions
+	for _, item := range r.offloading {
+		// keep items that are within limits or already have remote references
+		if r.stats.offloading.totalSizeOnDisk <= sizeLimit || item.remote != nil {
+			r.offloading[i] = item
+			i++
+			continue
+		}
+		evicted = append(evicted, item)
+		r.stats.offloading.Sub(item.instance.Info())
+	}
+
+	r.offloading = r.offloading[:i]
+	r.rebuildAllFractions()
+
+	return evicted
+}
+
 // PromoteToLocal moves fractions from sealing to local queue when sealing completes.
 // Maintains strict ordering - younger fractions wait for older ones to seal first.
 func (r *fractionRegistry) PromoteToLocal(active *activeProxy, sealed *frac.Sealed) {
@@ -306,6 +356,11 @@ func (r *fractionRegistry) removeFromOffloading(sealed *sealedProxy) {
 			count++
 		}
 	}
+
+	if count == len(r.offloading) { // not found to remove
+		return
+	}
+
 	r.offloading = r.offloading[:count]
 	r.stats.offloading.Sub(sealed.instance.Info())
 

fracmanager/lifecycle_manager.go

@@ -21,6 +21,7 @@ type lifecycleManager struct {
 	provider  *fractionProvider // Provider for fraction operations
 	flags     *StateManager     // Storage state flags
 	registry  *fractionRegistry // Fraction state registry
+	tasks     *TaskManager      // Background offloading tasks
 
 	sealingWg sync.WaitGroup
 }
@@ -36,16 +37,21 @@ func newLifecycleManager(
 		provider:  provider,
 		flags:     flags,
 		registry:  registry,
+		tasks:     NewTaskManager(),
 	}
 }
 
 // Maintain performs periodic lifecycle management tasks.
 // It is a CORE method of lifecycleManager
 // Coordinates rotation, offloading, cleanup based on configuration.
 func (lc *lifecycleManager) Maintain(ctx context.Context, config *Config, wg *sync.WaitGroup) {
-	lc.Rotate(config.FracSize, wg)
+	maxTotalSize := config.TotalSize + config.OffloadingQueueSize
+	lc.ManageRotation(config.FracSize, maxTotalSize, wg)
 	if config.OffloadingEnabled {
-		lc.OffloadLocal(ctx, config.TotalSize, wg)
+		lc.OffloadLocal(ctx, config.TotalSize, config.OffloadingRetryDelay, wg)
+		if config.OffloadingQueueSize > 0 {
+			lc.RemoveOverflowed(config.OffloadingQueueSize, wg)
+		}
 		lc.CleanRemote(config.OffloadingRetention, wg)
 	} else {
 		lc.CleanLocal(config.TotalSize, wg)
@@ -85,57 +91,95 @@ func (lc *lifecycleManager) Seal(active *activeProxy) error {
 	return nil
 }
 
-// RotateIfNeeded checks if active fraction needs rotation based on size limit
-// Creates new active fraction and starts sealing the previous one.
-func (lc *lifecycleManager) Rotate(sizeLimit uint64, wg *sync.WaitGroup) {
-	if lc.registry.Active().instance.Info().DocsOnDisk > sizeLimit {
-		active := lc.rotate()
-
-		wg.Add(1)
-		lc.sealingWg.Add(1)
-		go func() {
-			defer wg.Done()
-			defer lc.sealingWg.Done()
-			if err := lc.Seal(active); err != nil {
-				logger.Fatal("sealing error", zap.Error(err))
-			}
-		}()
+// ManageRotation checks if rotation is needed and manages the entire process
+// including suspension checks, fraction rotation, and asynchronous sealing.
+func (lc *lifecycleManager) ManageRotation(maxActiveSize, maxTotalSize uint64, wg *sync.WaitGroup) {
+	if maxTotalSize > 0 && lc.registry.SuspendIfOverCapacity(maxTotalSize) {
+		return
 	}
-}
 
-func (lc *lifecycleManager) rotate() *activeProxy {
-	active, err := lc.registry.Rotate(newActiveProxy(lc.provider.CreateActive()))
+	activeToSeal, err := lc.registry.RotateIfNeeded(maxActiveSize, func() *activeProxy {
+		return newActiveProxy(lc.provider.CreateActive())
+	})
 	if err != nil {
 		logger.Fatal("active fraction rotation error", zap.Error(err))
 	}
-	return active
+	if activeToSeal == nil {
+		return
+	}
+
+	wg.Add(1)
+	lc.sealingWg.Add(1)
+	go func() {
+		defer wg.Done()
+		defer lc.sealingWg.Done()
+		if err := lc.Seal(activeToSeal); err != nil {
+			logger.Fatal("sealing error", zap.Error(err))
+		}
+	}()
 }
 
 // OffloadLocal starts offloading of local fractions to remote storage
 // Selects fractions based on disk space usage and retention policy.
-func (lc *lifecycleManager) OffloadLocal(ctx context.Context, sizeLimit uint64, wg *sync.WaitGroup) {
+func (lc *lifecycleManager) OffloadLocal(ctx context.Context, sizeLimit uint64, retryDelay time.Duration, wg *sync.WaitGroup) {
 	toOffload, err := lc.registry.EvictLocal(true, sizeLimit)
 	if err != nil {
 		logger.Fatal("error releasing old fractions:", zap.Error(err))
 	}
 	for _, sealed := range toOffload {
 		wg.Add(1)
-		go func() {
+		lc.tasks.Run(sealed.instance.BaseFileName, ctx, func(ctx context.Context) {
 			defer wg.Done()
 
-			remote, _ := lc.TryOffload(ctx, sealed.instance)
+			remote := lc.OffloadWithRetry(ctx, sealed.instance, retryDelay)
+
 			lc.registry.PromoteToRemote(sealed, remote)
 
 			if remote == nil {
 				sealed.proxy.Redirect(emptyFraction{})
+				lc.infoCache.Remove(sealed.instance.Info().Name())
 			} else {
 				sealed.proxy.Redirect(remote)
 			}
 
 			// Free up local resources
 			sealed.instance.Suicide()
 			maintenanceTruncateTotal.Add(1)
-		}()
+		})
+	}
+}
+
+// OffloadWithRetry attempts to offload a fraction with retries until success or cancellation.
+// Returns the remote fraction instance and a boolean indicating whether offloading was not canceled.
+func (lc *lifecycleManager) OffloadWithRetry(ctx context.Context, sealed *frac.Sealed, retryDelay time.Duration) *frac.Remote {
+	start := time.Now()
+	for i := 0; ; i++ {
+		remote, err := lc.TryOffload(ctx, sealed)
+		if err == nil {
+			return remote
+		}
+
+		logger.Warn(
+			"fail to offload fraction",
+			zap.String("name", sealed.BaseFileName),
+			zap.Duration("offloading_time", time.Since(start)),
+			zap.Int("attempts", i),
+			zap.Error(err),
+		)
+
+		select {
+		case <-ctx.Done():
+			logger.Info(
+				"fraction offloading was stopped",
+				zap.String("name", sealed.BaseFileName),
+				zap.Duration("offloading_time", time.Since(start)),
+				zap.Int("attempts", i),
+				zap.Error(ctx.Err()),
+			)
+			return nil
+		case <-time.After(retryDelay):
+			// Wait before next retry attempt
+		}
 	}
 }
 
@@ -201,6 +245,21 @@ func (lc *lifecycleManager) CleanLocal(sizeLimit uint64, wg *sync.WaitGroup) {
 	}()
 }
 
+// RemoveOverflowed removes fractions from offloading queue that exceed size limit
+// Stops ongoing offloading tasks and cleans up both local and remote resources.
+func (lc *lifecycleManager) RemoveOverflowed(sizeLimit uint64, wg *sync.WaitGroup) {
+	evicted := lc.registry.EvictOverflowed(sizeLimit)
+	for _, item := range evicted {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			// Cancel the offloading task - this operation may take significant time
+			// hence executed in a separate goroutine to avoid blocking
+			lc.tasks.Cancel(item.instance.BaseFileName)
+		}()
+	}
+}
+
 // UpdateOldestMetric updates the prometheus metric with oldest fraction timestamp
 func (lc *lifecycleManager) UpdateOldestMetric() {
 	oldestFracTime.WithLabelValues("remote").Set((time.Duration(lc.registry.OldestTotal()) * time.Millisecond).Seconds())