address review

Signed-off-by: Jakub Sztandera <oss@kubuxu.com>

Author: Kubuxu

cmd/f3/observer.go

@@ -66,7 +66,7 @@ var observerCmd = cli.Command{
 			Usage: "The maximum length of time to keep the rotated files.",
 			Value: 2 * 7 * 24 * time.Hour,
 		},
-		&cli.Int64Flag{
+		&cli.Uint64Flag{
 			Name:  "retentionSize",
 			Usage: "The maximum size of the rotated files in megabytes. If not set, no limit is applied.",
 			Value: 0,
@@ -149,7 +149,7 @@ var observerCmd = cli.Command{
 			observer.WithMaxBatchSize(cctx.Int("maxBatchSize")),
 			observer.WithMaxBatchDelay(cctx.Duration("maxBatchDelay")),
 			observer.WithChainExchangeMaxMessageAge(cctx.Duration("chainExchangeMaxMessageAge")),
-			observer.WithMaxRetentionSize(cctx.Int64("retentionSize") * 1024 * 1024),
+			observer.WithMaxRetentionSize(cctx.Uint64("retentionSize") * 1024 * 1024),
 		}
 
 		var identity crypto.PrivKey

observer/observer.go

@@ -541,7 +541,7 @@ func (o *Observer) rotateMessages(ctx context.Context) error {
 				if err := os.Remove(filepath.Join(o.rotatePath, entry.Name())); err != nil {
 					logger.Errorw("Failed to remove retention policy for file", "file", entry.Name(), "err", err)
 				} else {
-					logger.Infow("Removed old file", "olderThan", o.retention, "file", entry.Name())
+					logger.Infow("Removed file due to time retention policy", "olderThan", o.retention, "file", entry.Name())
 				}
 			} else {
 				retainedSize += info.Size()
@@ -552,22 +552,25 @@ func (o *Observer) rotateMessages(ctx context.Context) error {
 
 	logger.Infow("Retention size", "retainedSize", retainedSize, "maxRetentionSize", o.maxRetentionSize)
 
-	if retainedSize > o.maxRetentionSize {
+	if o.maxRetentionSize > 0 && retainedSize > o.maxRetentionSize {
 		logger.Infow("Retention size exceeded, deleting oldest files", "retainedSize", retainedSize, "maxRetentionSize", o.maxRetentionSize)
 		// sort retained by modification time, oldest last
 		sort.Slice(retained, func(i, j int) bool {
 			return retained[i].ModTime().After(retained[j].ModTime())
 		})
 		// iterate in reverse order to delete oldest first
 		for i := len(retained) - 1; i >= 0; i-- {
+			fi := retained[i]
 			if retainedSize < o.maxRetentionSize {
 				break
 			}
-			if err := os.Remove(filepath.Join(o.rotatePath, retained[i].Name())); err != nil {
-				logger.Errorw("Failed to remove retention policy for file", "file", retained[i].Name(), "err", err)
+			if err := os.Remove(filepath.Join(o.rotatePath, fi.Name())); err != nil {
+				logger.Errorw("Failed to remove retention policy for file",
+					"file", fi.Name(), "err", err)
 			} else {
-				logger.Infow("Removed old file", "olderThan", o.retention, "file", retained[i].Name())
-				retainedSize -= retained[i].Size()
+				logger.Infow("Removed file due to size retention policy",
+					"size", fi.Size(), "file", fi.Name())
+				retainedSize -= fi.Size()
 				retained = retained[:i]
 			}
 		}

observer/options.go

@@ -3,6 +3,7 @@ package observer
 import (
 	"context"
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/filecoin-project/go-f3/blssig"
@@ -276,9 +277,12 @@ func WithRetention(retention time.Duration) Option {
 // This is weakly enforced, and the directory may grow larger than this
 // size. If the directory grows larger than this size, the oldest files
 // will be deleted until the directory size is below this size.
-func WithMaxRetentionSize(size int64) Option {
+func WithMaxRetentionSize(size uint64) Option {
 	return func(o *options) error {
-		o.maxRetentionSize = size
+		if size > math.MaxInt64 {
+			return fmt.Errorf("max retention size must be less than or equal to %d", math.MaxInt64)
+		}
+		o.maxRetentionSize = int64(size)
 		return nil
 	}
 }