LogPoller prune test

core/chains/evm/logpoller/log_poller.go

@@ -657,12 +657,18 @@
 				blockPruneTick = time.After(utils.WithJitter(lp.pollPeriod * 100))
 			}
 		case <-logPruneTick:
-			logPruneTick = time.After(utils.WithJitter(lp.pollPeriod * 2401)) // = 7^5 avoids common factors with 1000
-			if allRemoved, err := lp.PruneExpiredLogs(ctx); err != nil {
-				lp.lggr.Errorw("Unable to prune expired logs", "err", err)
-			} else if !allRemoved {
+			lp.lggr.Infof("Pruning LogPoller logs...")
+			logPruneTick = time.After(utils.WithJitter(5 * time.Minute)) // = 7^5 avoids common factors with 1000
+			allRemoved, err := lp.PruneExpiredLogs(ctx)
+			if allRemoved {
+				lp.lggr.Infof("Finished pruning LogPoller logs. Next pruning in ~ %s", lp.pollPeriod*2401)
+			} else {
 				// Tick faster when cleanup can't keep up with the pace of new logs
 				logPruneTick = time.After(utils.WithJitter(lp.pollPeriod * 241))
+				lp.lggr.Warnf("Partially finished pruning LogPoller logs. Will resume in ~ %s", lp.pollPeriod*241)
+			}
+			if err != nil {
+				lp.lggr.Errorw("Unable to prune expired logs", "err", err)
 			}
 		}
 	}
@@ -1092,8 +1098,16 @@
 // PruneExpiredLogs logs that are older than their retention period defined in Filter.
 // Returns whether all logs eligible for pruning were removed. If logPrunePageSize is set to 0, it will always return true.
 func (lp *logPoller) PruneExpiredLogs(ctx context.Context) (bool, error) {
-	rowsRemoved, err := lp.orm.DeleteExpiredLogs(ctx, lp.logPrunePageSize)
-	return lp.logPrunePageSize == 0 || rowsRemoved < lp.logPrunePageSize, err
+	rowsRemoved, err := lp.orm.DeleteExcessLogs(ctx, lp.logPrunePageSize)
+	if err != nil {
+		return false, err
+	}
+
+	rowsRemoved, err = lp.orm.DeleteExpiredLogs(ctx, lp.logPrunePageSize)
+	if err != nil || rowsRemoved < lp.logPrunePageSize {
+		return true, err
+	}
+	return lp.logPrunePageSize == 0 || err != nil || rowsRemoved < lp.logPrunePageSize, err
 }
 
 // Logs returns logs matching topics and address (exactly) in the given block range,

core/chains/evm/logpoller/observability.go

@@ -142,6 +142,12 @@ func (o *ObservedORM) DeleteExpiredLogs(ctx context.Context, limit int64) (int64
 	})
 }
 
+func (o *ObservedORM) DeleteExcessLogs(ctx context.Context, limit int64) (int64, error) {
+	return withObservedExecAndRowsAffected(o, "DeleteExcessLogs", del, func() (int64, error) {
+		return o.ORM.DeleteExcessLogs(ctx, limit)
+	})
+}
+
 func (o *ObservedORM) SelectBlockByNumber(ctx context.Context, n int64) (*LogPollerBlock, error) {
 	return withObservedQuery(o, "SelectBlockByNumber", func() (*LogPollerBlock, error) {
 		return o.ORM.SelectBlockByNumber(ctx, n)