code review feedback and delay pruning of activeset to next update

blocks/metrics.go

@@ -30,14 +30,6 @@ var (
 	failFetchCnt   = blockGenCount.WithLabelValues(failFetch)
 	failGenCnt     = blockGenCount.WithLabelValues(failGen)
 	failErrCnt     = blockGenCount.WithLabelValues(internalErr)
-
-	deleteLatency = metrics.NewHistogramWithBuckets(
-		"delete_duration",
-		namespace,
-		"duration in second to delete old proposals",
-		[]string{},
-		prometheus.ExponentialBuckets(0.01, 2, 10),
-	).WithLabelValues()
 )
 
 type collector struct {

mesh/metrics/prometheus.go

@@ -20,15 +20,4 @@ var (
 		[]string{},
 		prometheus.ExponentialBuckets(1, 2, 16),
 	)
-
-	pruneLatency = metrics.NewHistogramWithBuckets(
-		"prune_seconds",
-		Subsystem,
-		"prune time in seconds",
-		[]string{"step"},
-		prometheus.ExponentialBuckets(0.01, 2, 10),
-	)
-	PruneProposalLatency = pruneLatency.WithLabelValues("proposal")
-	PruneCertLatency     = pruneLatency.WithLabelValues("cert")
-	PrunePropTxLatency   = pruneLatency.WithLabelValues("proptxs")
 )

node/node.go

@@ -63,8 +63,10 @@ import (
 	"github.com/spacemeshos/go-spacemesh/p2p"
 	"github.com/spacemeshos/go-spacemesh/p2p/pubsub"
 	"github.com/spacemeshos/go-spacemesh/proposals"
+	"github.com/spacemeshos/go-spacemesh/prune"
 	"github.com/spacemeshos/go-spacemesh/signing"
 	"github.com/spacemeshos/go-spacemesh/sql"
+	"github.com/spacemeshos/go-spacemesh/sql/ballots/util"
 	"github.com/spacemeshos/go-spacemesh/sql/layers"
 	dbmetrics "github.com/spacemeshos/go-spacemesh/sql/metrics"
 	"github.com/spacemeshos/go-spacemesh/syncer"
@@ -645,7 +647,7 @@ func (app *App) initServices(ctx context.Context) error {
 	}
 
 	app.eg.Go(func() error {
-		mesh.Prune(ctx, mlog.Zap(), app.db, app.clock, app.Config.HareEligibility.ConfidenceParam, app.Config.DatabasePruneInterval)
+		prune.Prune(ctx, mlog.Zap(), app.db, app.clock, app.Config.Tortoise.Hdist, app.Config.DatabasePruneInterval)
 		return nil
 	})
 
@@ -1336,7 +1338,7 @@ func (app *App) setupDBs(ctx context.Context, lg log.Log) error {
 	sqlDB, err := sql.Open("file:"+filepath.Join(dbPath, dbFile),
 		sql.WithConnections(app.Config.DatabaseConnections),
 		sql.WithLatencyMetering(app.Config.DatabaseLatencyMetering),
-		sql.WithV4PreMigration(mesh.ExtractActiveSet),
+		sql.WithV4Migration(util.ExtractActiveSet),
 	)
 	if err != nil {
 		return fmt.Errorf("open sqlite db %w", err)

prune/interface.go

@@ -0,0 +1,11 @@
+package prune
+
+import (
+	"github.com/spacemeshos/go-spacemesh/common/types"
+)
+
+//go:generate mockgen -typed -package=prune -destination=./mocks.go -source=./interface.go
+
+type layerClock interface {
+	CurrentLayer() types.LayerID
+}

prune/metrics.go

@@ -0,0 +1,22 @@
+package prune
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/spacemeshos/go-spacemesh/metrics"
+)
+
+const namespace = "prune"
+
+var (
+	pruneLatency = metrics.NewHistogramWithBuckets(
+		"prune_seconds",
+		namespace,
+		"prune time in seconds",
+		[]string{"step"},
+		prometheus.ExponentialBuckets(0.01, 2, 10),
+	)
+	proposalLatency = pruneLatency.WithLabelValues("proposal")
+	certLatency     = pruneLatency.WithLabelValues("cert")
+	propTxLatency   = pruneLatency.WithLabelValues("proptxs")
+)

prune/prune.go

@@ -0,0 +1,60 @@
+package prune
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/spacemeshos/go-spacemesh/common/types"
+	"github.com/spacemeshos/go-spacemesh/sql"
+	"github.com/spacemeshos/go-spacemesh/sql/certificates"
+	"github.com/spacemeshos/go-spacemesh/sql/proposals"
+	"github.com/spacemeshos/go-spacemesh/sql/transactions"
+)
+
+func Prune(
+	ctx context.Context,
+	logger *zap.Logger,
+	db sql.Executor,
+	lc layerClock,
+	safeDist uint32,
+	interval time.Duration,
+) {
+	logger.With().Info("db pruning launched",
+		zap.Uint32("dist", safeDist),
+		zap.Duration("interval", interval),
+	)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(interval):
+			oldest := lc.CurrentLayer() - types.LayerID(safeDist)
+			t0 := time.Now()
+			if err := proposals.DeleteBefore(db, oldest); err != nil {
+				logger.Error("failed to delete proposals",
+					zap.Stringer("lid", oldest),
+					zap.Error(err),
+				)
+			}
+			proposalLatency.Observe(time.Since(t0).Seconds())
+			t1 := time.Now()
+			if err := certificates.DeleteCertBefore(db, oldest); err != nil {
+				logger.Error("failed to delete certificates",
+					zap.Stringer("lid", oldest),
+					zap.Error(err),
+				)
+			}
+			certLatency.Observe(time.Since(t1).Seconds())
+			t2 := time.Now()
+			if err := transactions.DeleteProposalTxsBefore(db, oldest); err != nil {
+				logger.Error("failed to delete proposal tx mapping",
+					zap.Stringer("lid", oldest),
+					zap.Error(err),
+				)
+			}
+			propTxLatency.Observe(time.Since(t2).Seconds())
+		}
+	}
+}