feat: save expired batches in db

pkg/storer/reserve.go

@@ -9,6 +9,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"path"
 	"sync"
 	"time"
 
@@ -24,7 +25,6 @@ const (
 	reserveUnreserved    = "reserveUnreserved"
 	reserveUpdateLockKey = "reserveUpdateLockKey"
 	batchExpiry          = "batchExpiry"
-	expiredBatchAccess   = "expiredBatchAccess"
 
 	cleanupDur = time.Hour * 6
 )
@@ -123,14 +123,20 @@ func (db *DB) evictionWorker(ctx context.Context) {
 	defer overCapUnsub()
 
 	cleanupExpired := func() {
-		db.lock.Lock(expiredBatchAccess)
-		batchesToEvict := make([][]byte, len(db.expiredBatches))
-		copy(batchesToEvict, db.expiredBatches)
-		db.expiredBatches = nil
-		db.lock.Unlock(expiredBatchAccess)
-
 		defer db.events.Trigger(reserveUnreserved)
 
+		batchesToEvict := make([][]byte, 0)
+		err := db.repo.IndexStore().Iterate(storage.Query{
+			Factory:      func() storage.Item { return new(expiredBatchItem) },
+			ItemProperty: storage.QueryItemID,
+		}, func(result storage.Result) (bool, error) {
+			batchesToEvict = append(batchesToEvict, []byte(result.ID))
+			return false, nil
+		})
+		if err != nil {
+			db.logger.Error(err, "iterate expired batches")
+		}
+
 		if len(batchesToEvict) > 0 {
 			for _, batchID := range batchesToEvict {
 				evicted, err := db.evictBatch(ctx, batchID, swarm.MaxBins)
@@ -143,12 +149,22 @@ func (db *DB) evictionWorker(ctx context.Context) {
 						"batch", hex.EncodeToString(batchID),
 						"total_evicted", evicted,
 					)
+
+					err = db.Execute(ctx, func(tx internal.Storage) error {
+						return tx.IndexStore().Delete(&expiredBatchItem{BatchID: batchID})
+					})
+					if err != nil {
+						db.logger.Error(err, "delete expired batch", "batch", hex.EncodeToString(batchID))
+					}
 				}
 				db.metrics.ExpiredBatchCount.Inc()
 			}
 		}
 	}
 
+	// initial cleanup
+	cleanupExpired()
+
 	time.AfterFunc(30*time.Minute, func() {
 		db.logger.Info("initial reserve cleanup started")
 		if err := db.reserveCleanup(ctx); err != nil {
@@ -178,7 +194,7 @@ func (db *DB) evictionWorker(ctx context.Context) {
 			cleanupExpired()
 
 			if err := db.reserveCleanup(ctx); err != nil {
-				db.logger.Error(err, "cleanup")
+				db.logger.Error(err, "reserve cleanup")
 			}
 		}
 	}
@@ -270,19 +286,54 @@ func (db *DB) ReservePutter() storage.Putter {
 	}
 }
 
+type expiredBatchItem struct {
+	BatchID []byte
+}
+
+func (e *expiredBatchItem) ID() string {
+	return string(e.BatchID)
+}
+
+func (e *expiredBatchItem) Namespace() string {
+	return "expiredBatchItem"
+}
+
+func (e *expiredBatchItem) Marshal() ([]byte, error) {
+	return nil, nil
+}
+
+func (e *expiredBatchItem) Unmarshal(_ []byte) error {
+	return nil
+}
+
+func (e *expiredBatchItem) Clone() storage.Item {
+	if e == nil {
+		return nil
+	}
+	return &expiredBatchItem{
+		BatchID: append([]byte(nil), e.BatchID...),
+	}
+}
+
+func (e *expiredBatchItem) String() string {
+	return path.Join(e.Namespace(), e.ID())
+}
+
 // EvictBatch evicts all chunks belonging to a batch from the reserve.
-func (db *DB) EvictBatch(ctx context.Context, batchID []byte) (err error) {
+func (db *DB) EvictBatch(ctx context.Context, batchID []byte) error {
 	if db.reserve == nil {
 		// if reserve is not configured, do nothing
 		return nil
 	}
 
-	db.lock.Lock(expiredBatchAccess)
-	db.expiredBatches = append(db.expiredBatches, batchID)
-	db.lock.Unlock(expiredBatchAccess)
+	err := db.Execute(ctx, func(tx internal.Storage) error {
+		return tx.IndexStore().Put(&expiredBatchItem{BatchID: batchID})
+	})
+	if err != nil {
+		return fmt.Errorf("save expired batch: %w", err)
+	}
 
 	db.events.Trigger(batchExpiry)
-
 	return nil
 }
 

pkg/storer/storer.go

@@ -473,7 +473,6 @@ type DB struct {
 	setSyncerOnce    sync.Once
 	syncer           Syncer
 	opts             workerOpts
-	expiredBatches   [][]byte
 }
 
 type workerOpts struct {