storagenode/collector: reverse flat file store deletion order

We've killed many times the expiration jobs that were running the old
implementation, which left a lot of hour files that have already been
processed. Running the jobs again requires them to process these files
again, which significantly delays the deletion of the remaining expired
pieces.

By reversing the order, the job will priorities the expired pieces that
has not been deleted yet.

Change-Id: I95eaafe8ecc603fd939a34f91467b98766b715dd

Author: kaloyan-raev

storagenode/collector/service.go

@@ -25,6 +25,7 @@ type Config struct {
 	ExpirationGracePeriod time.Duration `help:"how long should the collector wait before deleting expired pieces. Should not be less than 30 min since nodes are allowed to be 30 mins out of sync with the satellite." default:"1h0m0s"`
 	ExpirationBatchSize   int           `help:"how many expired pieces to delete in one batch. If <= 0, all expired pieces will be deleted in one batch. (ignored by flat file store)" default:"1000"`
 	FlatFileBatchLimit    int           `help:"how many per hour flat files can be deleted in one batch." default:"5"`
+	ReverseOrder          bool          `help:"delete expired pieces in reverse order (recently expired first)" default:"false"`
 }
 
 // Service implements collecting expired pieces on the storage node.
@@ -37,7 +38,7 @@ type Service struct {
 
 	Loop *sync2.Cycle
 
-	limits                pieces.ExpirationLimits
+	opts                  pieces.ExpirationOptions
 	expirationGracePeriod time.Duration
 }
 
@@ -48,14 +49,16 @@ func NewService(log *zap.Logger, pieceStore *pieces.Store, usedSerials *usedseri
 		config.ExpirationGracePeriod = 1 * time.Hour
 	}
 
+	opts := pieces.DefaultExpirationOptions()
+	opts.Limits.BatchSize = config.ExpirationBatchSize
+	opts.Limits.FlatFileLimit = config.FlatFileBatchLimit
+	opts.ReverseOrder = config.ReverseOrder
+
 	return &Service{
-		log:         log,
-		pieces:      pieceStore,
-		usedSerials: usedSerials,
-		limits: pieces.ExpirationLimits{
-			BatchSize:     config.ExpirationBatchSize,
-			FlatFileLimit: config.FlatFileBatchLimit,
-		},
+		log:                   log,
+		pieces:                pieceStore,
+		usedSerials:           usedSerials,
+		opts:                  opts,
 		expirationGracePeriod: config.ExpirationGracePeriod,
 		Loop:                  sync2.NewCycle(config.Interval),
 	}
@@ -100,7 +103,7 @@ func (service *Service) Collect(ctx context.Context, now time.Time) (err error)
 	}()
 
 	for {
-		infoLists, err := service.pieces.GetExpiredBatchSkipV0(ctx, now, service.limits)
+		infoLists, err := service.pieces.GetExpiredBatchSkipV0(ctx, now, service.opts)
 		if err != nil {
 			return errs.Wrap(err)
 		}
@@ -128,7 +131,7 @@ func (service *Service) Collect(ctx context.Context, now time.Time) (err error)
 		}
 
 		// delete the batch from the database
-		if deleteErr := service.pieces.DeleteExpiredBatchSkipV0(ctx, now, service.limits); deleteErr != nil {
+		if deleteErr := service.pieces.DeleteExpiredBatchSkipV0(ctx, now, service.opts); deleteErr != nil {
 			service.log.Error("error during deleting expired pieces: ", zap.Error(deleteErr))
 			return errs.Wrap(deleteErr)
 		}

storagenode/pieces/combined_expiration_store.go

@@ -36,15 +36,15 @@ func (c *CombinedExpirationStore) SetExpiration(ctx context.Context, satellite s
 }
 
 // GetExpired returns the expired pieces.
-func (c *CombinedExpirationStore) GetExpired(ctx context.Context, expiresBefore time.Time, limits ExpirationLimits) ([]*ExpiredInfoRecords, error) {
+func (c *CombinedExpirationStore) GetExpired(ctx context.Context, expiresBefore time.Time, opts ExpirationOptions) ([]*ExpiredInfoRecords, error) {
 	var errList errs.Group
-	expired, err := c.flatFileStore.GetExpired(ctx, expiresBefore, limits)
+	expired, err := c.flatFileStore.GetExpired(ctx, expiresBefore, opts)
 	if err != nil {
 		errList.Add(err)
 	}
 
 	if c.chainedDB != nil {
-		chainedExpired, err := c.chainedDB.GetExpired(ctx, expiresBefore, limits)
+		chainedExpired, err := c.chainedDB.GetExpired(ctx, expiresBefore, opts)
 		if err != nil {
 			errList.Add(err)
 		}
@@ -72,17 +72,17 @@ func (c *CombinedExpirationStore) DeleteExpirations(ctx context.Context, expires
 }
 
 // DeleteExpirationsBatch deletes the expirations for the given time.
-func (c *CombinedExpirationStore) DeleteExpirationsBatch(ctx context.Context, now time.Time, limits ExpirationLimits) error {
+func (c *CombinedExpirationStore) DeleteExpirationsBatch(ctx context.Context, now time.Time, opts ExpirationOptions) error {
 	var errList errs.Group
 
-	c.log.Debug("deleting expired pieces from flat file store", zap.Time("expiresAt", now), zap.Any("limits", limits))
-	if err := c.flatFileStore.DeleteExpirationsBatch(ctx, now, limits); err != nil {
+	c.log.Debug("deleting expired pieces from flat file store", zap.Time("expiresAt", now), zap.Any("opts", opts))
+	if err := c.flatFileStore.DeleteExpirationsBatch(ctx, now, opts); err != nil {
 		errList.Add(err)
 	}
 
 	if c.chainedDB != nil {
-		c.log.Debug("deleting expired pieces from flat file store", zap.Time("expiresAt", now), zap.Any("limits", limits))
-		errList.Add(c.chainedDB.DeleteExpirationsBatch(ctx, now, limits))
+		c.log.Debug("deleting expired pieces from flat file store", zap.Time("expiresAt", now), zap.Any("opts", opts))
+		errList.Add(c.chainedDB.DeleteExpirationsBatch(ctx, now, opts))
 	}
 
 	return errList.Err()

storagenode/pieces/combined_expiration_store_test.go

@@ -50,7 +50,7 @@ func TestPieceExpirationCombinedStore(t *testing.T) {
 		require.NoError(t, err)
 
 		// check to see that values are in both backends
-		expirationLists, err := combinedStore.GetExpired(ctx, now.Add(72*time.Hour), pieces.DefaultExpirationLimits())
+		expirationLists, err := combinedStore.GetExpired(ctx, now.Add(72*time.Hour), pieces.DefaultExpirationOptions())
 		require.NoError(t, err)
 		expirationInfos := pieces.FlattenExpirationInfoLists(expirationLists)
 		require.Len(t, expirationInfos, 4)
@@ -71,14 +71,13 @@ func TestPieceExpirationCombinedStore(t *testing.T) {
 		})
 
 		// delete up to now+36h
-		err = combinedStore.DeleteExpirationsBatch(ctx, now.Add(36*time.Hour), pieces.ExpirationLimits{
-			BatchSize:     10,
-			FlatFileLimit: -1,
-		})
+		opts := pieces.DefaultExpirationOptions()
+		opts.Limits.BatchSize = 10
+		err = combinedStore.DeleteExpirationsBatch(ctx, now.Add(36*time.Hour), opts)
 		require.NoError(t, err)
 
 		// piece1 should be deleted from both databases, and not the others
-		expirationLists, err = combinedStore.GetExpired(ctx, now.Add(72*time.Hour), pieces.DefaultExpirationLimits())
+		expirationLists, err = combinedStore.GetExpired(ctx, now.Add(72*time.Hour), pieces.DefaultExpirationOptions())
 		require.NoError(t, err)
 		expirationInfos = pieces.FlattenExpirationInfoLists(expirationLists)
 		require.Len(t, expirationInfos, 2)
@@ -93,7 +92,7 @@ func TestPieceExpirationCombinedStore(t *testing.T) {
 		})
 
 		// querying sqlite3 db only
-		expirationLists, err = sqliteDB.GetExpired(ctx, now.Add(72*time.Hour), pieces.DefaultExpirationLimits())
+		expirationLists, err = sqliteDB.GetExpired(ctx, now.Add(72*time.Hour), pieces.DefaultExpirationOptions())
 		require.NoError(t, err)
 		expirationInfos = pieces.FlattenExpirationInfoLists(expirationLists)
 		require.Len(t, expirationInfos, 1)

storagenode/pieces/pieceexpiration.go

@@ -136,7 +136,7 @@ var monGetExpired = mon.Task()
 // GetExpired gets piece IDs that expire or have expired before the given time.
 // It returns the lastExpiration time, which can be used to delete all expirations
 // up to that time.
-func (peStore *PieceExpirationStore) GetExpired(ctx context.Context, now time.Time, limits ExpirationLimits) (infos []*ExpiredInfoRecords, err error) {
+func (peStore *PieceExpirationStore) GetExpired(ctx context.Context, now time.Time, opts ExpirationOptions) (infos []*ExpiredInfoRecords, err error) {
 	defer monGetExpired(&ctx)(&err)
 
 	satellites, err := peStore.getSatellitesWithExpirations(ctx)
@@ -145,7 +145,7 @@ func (peStore *PieceExpirationStore) GetExpired(ctx context.Context, now time.Ti
 	}
 	var errList errs.Group
 	for _, satelliteID := range satellites {
-		satelliteInfos, err := peStore.GetExpiredForSatellite(ctx, satelliteID, now, limits)
+		satelliteInfos, err := peStore.GetExpiredForSatellite(ctx, satelliteID, now, opts)
 		if err != nil {
 			errList.Add(ErrPieceExpiration.Wrap(err))
 		}
@@ -161,13 +161,13 @@ var monDeleteExpirations = mon.Task()
 func (peStore *PieceExpirationStore) DeleteExpirations(ctx context.Context, expiresAt time.Time) (err error) {
 	defer monDeleteExpirations(&ctx)(&err)
 
-	errList := peStore.deleteExpirationsFromAllSatellites(ctx, expiresAt, -1)
+	errList := peStore.deleteExpirationsFromAllSatellites(ctx, expiresAt, DefaultExpirationOptions())
 	return errList.Err()
 }
 
 var monDeleteExpirationsFromAllSatellites = mon.Task()
 
-func (peStore *PieceExpirationStore) deleteExpirationsFromAllSatellites(ctx context.Context, expiresAt time.Time, numFiles int) (errList errs.Group) {
+func (peStore *PieceExpirationStore) deleteExpirationsFromAllSatellites(ctx context.Context, expiresAt time.Time, opts ExpirationOptions) (errList errs.Group) {
 	defer monDeleteExpirationsFromAllSatellites(&ctx)(nil)
 
 	satellites, err := peStore.getSatellitesWithExpirations(ctx)
@@ -176,7 +176,7 @@ func (peStore *PieceExpirationStore) deleteExpirationsFromAllSatellites(ctx cont
 		return errList
 	}
 	for _, satelliteID := range satellites {
-		err := peStore.DeleteExpirationsForSatellite(ctx, satelliteID, expiresAt, numFiles)
+		err := peStore.DeleteExpirationsForSatellite(ctx, satelliteID, expiresAt, opts)
 		errList.Add(ErrPieceExpiration.Wrap(err))
 	}
 	return errList
@@ -218,10 +218,10 @@ var monGetExpiredForSatellite = mon.Task()
 
 // GetExpiredForSatellite gets piece IDs that expire or have expired before the
 // given time for a specific satellite.
-func (peStore *PieceExpirationStore) GetExpiredForSatellite(ctx context.Context, satellite storj.NodeID, now time.Time, limits ExpirationLimits) (infos *ExpiredInfoRecords, err error) {
+func (peStore *PieceExpirationStore) GetExpiredForSatellite(ctx context.Context, satellite storj.NodeID, now time.Time, opts ExpirationOptions) (infos *ExpiredInfoRecords, err error) {
 	defer monGetExpiredForSatellite(&ctx)(&err)
 
-	elapsed, err := peStore.getElapsedHoursWithExpirations(ctx, satellite, now, limits.FlatFileLimit)
+	elapsed, err := peStore.getElapsedHoursWithExpirations(ctx, satellite, now, opts)
 	if err != nil {
 		return nil, ErrPieceExpiration.Wrap(err)
 	}
@@ -327,10 +327,10 @@ var monDeleteExpirationsForSatellite = mon.Task()
 
 // DeleteExpirationsForSatellite deletes information about piece expirations
 // before the given time for a specific satellite.
-func (peStore *PieceExpirationStore) DeleteExpirationsForSatellite(ctx context.Context, satellite storj.NodeID, now time.Time, numFiles int) (err error) {
+func (peStore *PieceExpirationStore) DeleteExpirationsForSatellite(ctx context.Context, satellite storj.NodeID, now time.Time, opts ExpirationOptions) (err error) {
 	defer monDeleteExpirationsForSatellite(&ctx)(&err)
 
-	elapsed, err := peStore.getElapsedHoursWithExpirations(ctx, satellite, now, numFiles)
+	elapsed, err := peStore.getElapsedHoursWithExpirations(ctx, satellite, now, opts)
 	if err != nil {
 		return ErrPieceExpiration.Wrap(err)
 	}
@@ -358,14 +358,14 @@ func (peStore *PieceExpirationStore) DeleteExpirationsForSatellite(ctx context.C
 }
 
 // DeleteExpirationsBatch removes expiration records for pieces that have expired before the given time.
-func (peStore *PieceExpirationStore) DeleteExpirationsBatch(ctx context.Context, now time.Time, limits ExpirationLimits) error {
-	errList := peStore.deleteExpirationsFromAllSatellites(ctx, now, limits.FlatFileLimit)
+func (peStore *PieceExpirationStore) DeleteExpirationsBatch(ctx context.Context, now time.Time, opts ExpirationOptions) error {
+	errList := peStore.deleteExpirationsFromAllSatellites(ctx, now, opts)
 	return errList.Err()
 }
 
 var monGetElapsedHoursWithExpirations = mon.Task()
 
-func (peStore *PieceExpirationStore) getElapsedHoursWithExpirations(ctx context.Context, satellite storj.NodeID, now time.Time, numOfFiles int) (elapsed []time.Time, err error) {
+func (peStore *PieceExpirationStore) getElapsedHoursWithExpirations(ctx context.Context, satellite storj.NodeID, now time.Time, opts ExpirationOptions) (elapsed []time.Time, err error) {
 	defer monGetElapsedHoursWithExpirations(&ctx)(&err)
 
 	satelliteDir := PathEncoding.EncodeToString(satellite[:])
@@ -382,6 +382,9 @@ func (peStore *PieceExpirationStore) getElapsedHoursWithExpirations(ctx context.
 		return nil, ErrPieceExpiration.Wrap(err)
 	}
 	slices.Sort(names)
+	if opts.ReverseOrder {
+		slices.Reverse(names)
+	}
 	for _, name := range names {
 		hour, err := time.ParseInLocation(PieceExpirationFileNameFormat, name, time.UTC)
 		if err != nil {
@@ -394,7 +397,7 @@ func (peStore *PieceExpirationStore) getElapsedHoursWithExpirations(ctx context.
 		}
 
 		// if we have a limit, we can stop once we have enough elapsed hours
-		if numOfFiles > 0 && len(elapsed) >= numOfFiles {
+		if opts.Limits.FlatFileLimit > 0 && len(elapsed) >= opts.Limits.FlatFileLimit {
 			break
 		}
 	}

storagenode/pieces/pieceexpiration_test.go

@@ -27,7 +27,7 @@ func PieceExpirationFunctionalityTest(ctx context.Context, t *testing.T, expireD
 	pieceID := testrand.PieceID()
 
 	// GetExpired with no matches
-	expiredLists, err := expireDB.GetExpired(ctx, time.Now(), DefaultExpirationLimits())
+	expiredLists, err := expireDB.GetExpired(ctx, time.Now(), DefaultExpirationOptions())
 	require.NoError(t, err)
 	expired := FlattenExpirationInfoLists(expiredLists)
 	require.Len(t, expired, 0)
@@ -47,7 +47,7 @@ func PieceExpirationFunctionalityTest(ctx context.Context, t *testing.T, expireD
 	require.NoError(t, err)
 
 	// GetExpired normal usage
-	expiredLists, err = expireDB.GetExpired(ctx, expireAt, DefaultExpirationLimits())
+	expiredLists, err = expireDB.GetExpired(ctx, expireAt, DefaultExpirationOptions())
 	require.NoError(t, err)
 	expired = FlattenExpirationInfoLists(expiredLists)
 	require.Len(t, expired, 1)
@@ -57,7 +57,7 @@ func PieceExpirationFunctionalityTest(ctx context.Context, t *testing.T, expireD
 	require.NoError(t, err)
 
 	// Should not be there anymore
-	expiredLists, err = expireDB.GetExpired(ctx, expireAt.Add(365*24*time.Hour), DefaultExpirationLimits())
+	expiredLists, err = expireDB.GetExpired(ctx, expireAt.Add(365*24*time.Hour), DefaultExpirationOptions())
 	expired = FlattenExpirationInfoLists(expiredLists)
 	require.NoError(t, err)
 	require.Len(t, expired, 0)
@@ -112,7 +112,7 @@ func TestPieceExpirationStoreInDepth(t *testing.T) {
 	}
 
 	afterExpirations := now.Add(time.Duration(numSatellites*numPieces+1) * time.Hour)
-	expirationLists, err := store.GetExpired(ctx, afterExpirations, DefaultExpirationLimits())
+	expirationLists, err := store.GetExpired(ctx, afterExpirations, DefaultExpirationOptions())
 	require.NoError(t, err)
 
 	require.Len(t, expirationLists, numSatellites)
@@ -136,10 +136,10 @@ func TestPieceExpirationStoreInDepth(t *testing.T) {
 		}
 	}
 
-	err = store.DeleteExpirationsForSatellite(ctx, satellites[0], afterExpirations, -1)
+	err = store.DeleteExpirationsForSatellite(ctx, satellites[0], afterExpirations, DefaultExpirationOptions())
 	require.NoError(t, err)
 
-	expirationLists, err = store.GetExpired(ctx, afterExpirations, DefaultExpirationLimits())
+	expirationLists, err = store.GetExpired(ctx, afterExpirations, DefaultExpirationOptions())
 	require.NoError(t, err)
 	expirationInfos := FlattenExpirationInfoLists(expirationLists)
 	require.Len(t, expirationInfos, (numSatellites-1)*numPieces)
@@ -157,7 +157,7 @@ func TestPieceExpirationStoreInDepth(t *testing.T) {
 	err = store.DeleteExpirations(ctx, afterExpirations)
 	require.NoError(t, err)
 
-	expirationLists, err = store.GetExpired(ctx, afterExpirations, DefaultExpirationLimits())
+	expirationLists, err = store.GetExpired(ctx, afterExpirations, DefaultExpirationOptions())
 	require.NoError(t, err)
 	expirationInfos = FlattenExpirationInfoLists(expirationLists)
 
@@ -208,7 +208,7 @@ func TestPieceExpirationStoreFileTruncation(t *testing.T) {
 	require.NoError(t, f.Close())
 
 	// check all piece expirations in store
-	expirationLists, err := store.GetExpired(ctx, now.Add(time.Hour), DefaultExpirationLimits())
+	expirationLists, err := store.GetExpired(ctx, now.Add(time.Hour), DefaultExpirationOptions())
 	require.NoError(t, err)
 	require.Len(t, expirationLists, 1)
 	require.Equal(t, satelliteID, expirationLists[0].SatelliteID)
@@ -222,7 +222,7 @@ func TestPieceExpirationStoreFileTruncation(t *testing.T) {
 	err = store.SetExpiration(ctx, satelliteID, pieceID2, now, 200)
 	require.NoError(t, err)
 
-	expirationLists, err = store.GetExpired(ctx, now.Add(time.Hour), DefaultExpirationLimits())
+	expirationLists, err = store.GetExpired(ctx, now.Add(time.Hour), DefaultExpirationOptions())
 	require.NoError(t, err)
 	expirationInfos := FlattenExpirationInfoLists(expirationLists)
 	require.Len(t, expirationInfos, 2)