Add a log debouncer to op-service.log package (#259)

QuentinI · dailinsubjam · commit 73cd70b0ab41 · 2025-11-08T00:52:06.000Z
(cherry picked from commit 788d28f)
diff --git a/op-service/log/cli.go b/op-service/log/cli.go
@@ -234,7 +234,8 @@ func NewLogHandler(wr io.Writer, cfg CLIConfig) slog.Handler {
 // The log handler of the logger is a LvlSetter, i.e. the log level can be changed as needed.
 func NewLogger(wr io.Writer, cfg CLIConfig) log.Logger {
 	h := NewLogHandler(wr, cfg)
-	l := log.NewLogger(h)
+	debounced := NewDebouncingHandler(h)
+	l := log.NewLogger(debounced)
 	if cfg.Pid {
 		l = l.With("pid", os.Getpid())
 	}
@@ -247,7 +248,8 @@ func NewLogger(wr io.Writer, cfg CLIConfig) log.Logger {
 // Geth and other components may use the global logger however,
 // and it is thus recommended to set the global log handler to catch these logs.
 func SetGlobalLogHandler(h slog.Handler) {
-	l := log.NewLogger(h)
+	debounced := NewDebouncingHandler(h)
+	l := log.NewLogger(debounced)
 	ctx := logfilter.AddLogAttrToContext(context.Background(), "global", true)
 	l.SetContext(ctx)
 	log.SetDefault(l)
diff --git a/op-service/log/debouncer.go b/op-service/log/debouncer.go
@@ -0,0 +1,74 @@
+package log
+
+import (
+	"context"
+	"log/slog"
+	"sync/atomic"
+	"time"
+
+	lru "github.com/hashicorp/golang-lru/v2"
+)
+
+const (
+	// DebounceDuration is the time window during which duplicate messages are suppressed
+	DebounceDuration = 100 * time.Millisecond
+	// DebounceTickerInterval is how often we check and report debounced message counts
+	DebounceTickerInterval = 5 * time.Second
+	// DebounceWarningMessage is the message logged when messages have been debounced
+	DebounceWarningMessage = "Some messages were debounced"
+)
+
+type DebounchingHandler struct {
+	handler  slog.Handler
+	messages *lru.Cache[string, time.Time]
+	counter  atomic.Uint64
+	ticker   *time.Ticker
+}
+
+func NewDebouncingHandler(handler slog.Handler) *DebounchingHandler {
+	messages, _ := lru.New[string, time.Time](1024)
+	return &DebounchingHandler{
+		handler:  handler,
+		messages: messages,
+		ticker:   time.NewTicker(DebounceTickerInterval),
+	}
+}
+
+func (h *DebounchingHandler) Enabled(ctx context.Context, lvl slog.Level) bool {
+	return h.handler.Enabled(ctx, lvl)
+}
+
+func (h *DebounchingHandler) Handle(ctx context.Context, record slog.Record) error {
+	select {
+	case <-h.ticker.C:
+		cntr := h.counter.Load()
+		h.counter.Store(0)
+
+		if cntr > 0 {
+			warningRecord := slog.NewRecord(time.Now(), slog.LevelWarn, DebounceWarningMessage, 0)
+			warningRecord.Add("nDebounced", cntr)
+			err := h.handler.Handle(ctx, warningRecord)
+			if err != nil {
+				return err
+			}
+		}
+
+	default:
+	}
+
+	if last, ok := h.messages.Get(record.Message); ok && time.Since(last) < DebounceDuration {
+		h.counter.Add(1)
+		return nil
+	}
+	h.messages.Add(record.Message, time.Now())
+
+	return h.handler.Handle(ctx, record)
+}
+
+func (h *DebounchingHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
+	return NewDebouncingHandler(h.handler.WithAttrs(attrs))
+}
+
+func (h *DebounchingHandler) WithGroup(name string) slog.Handler {
+	return NewDebouncingHandler(h.handler.WithGroup(name))
+}
diff --git a/op-service/log/debouncer_test.go b/op-service/log/debouncer_test.go
@@ -0,0 +1,255 @@
+package log
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum/go-ethereum/log"
+)
+
+// safeTestRecorder is a thread-safe version of testRecorder for concurrent tests
+type safeTestRecorder struct {
+	mu      sync.Mutex
+	records []slog.Record
+}
+
+func (r *safeTestRecorder) Enabled(context.Context, slog.Level) bool {
+	return true
+}
+
+func (r *safeTestRecorder) Handle(_ context.Context, rec slog.Record) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.records = append(r.records, rec)
+	return nil
+}
+
+func (r *safeTestRecorder) WithAttrs([]slog.Attr) slog.Handler { return r }
+func (r *safeTestRecorder) WithGroup(string) slog.Handler      { return r }
+
+func (r *safeTestRecorder) GetRecords() []slog.Record {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	// Return a copy to avoid race conditions
+	result := make([]slog.Record, len(r.records))
+	copy(result, r.records)
+	return result
+}
+
+func (r *safeTestRecorder) Len() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.records)
+}
+
+func TestDebouncingHandler_Basic(t *testing.T) {
+	h := new(testRecorder)
+	d := NewDebouncingHandler(h)
+	logger := log.NewLogger(d)
+
+	// First message should go through
+	logger.Info("hello world")
+	require.Len(t, h.records, 1)
+	require.Equal(t, "hello world", h.records[0].Message)
+
+	// Same message within 100ms should be dropped
+	logger.Info("hello world")
+	require.Len(t, h.records, 1)
+
+	// Different message should go through
+	logger.Info("different message")
+	require.Len(t, h.records, 2)
+	require.Equal(t, "different message", h.records[1].Message)
+
+	// Wait for debounce period to expire
+	time.Sleep(DebounceDuration + 1*time.Millisecond)
+
+	// Same message should now go through again
+	logger.Info("hello world")
+	require.Len(t, h.records, 3)
+	require.Equal(t, "hello world", h.records[2].Message)
+}
+
+func TestDebouncingHandler_MultipleMessages(t *testing.T) {
+	h := new(testRecorder)
+	d := NewDebouncingHandler(h)
+	logger := log.NewLogger(d)
+
+	// Send multiple different messages
+	messages := []string{"msg1", "msg2", "msg3", "msg4", "msg5"}
+	for _, msg := range messages {
+		logger.Info(msg)
+	}
+	require.Len(t, h.records, len(messages))
+
+	// Try to resend them immediately - all should be dropped
+	for _, msg := range messages {
+		logger.Info(msg)
+	}
+	require.Len(t, h.records, len(messages))
+
+	// Wait for debounce period
+	time.Sleep(DebounceDuration + 1*time.Millisecond)
+
+	// Now they should all go through again
+	for _, msg := range messages {
+		logger.Info(msg)
+	}
+	require.Len(t, h.records, 2*len(messages))
+}
+
+func TestDebouncingHandler_CacheEviction(t *testing.T) {
+	h := new(testRecorder)
+	d := NewDebouncingHandler(h)
+	logger := log.NewLogger(d)
+
+	// Generate more than 1024 unique messages to trigger LRU eviction
+	const numMessages = 1100
+	for i := range numMessages {
+		logger.Info(slog.IntValue(i).String())
+	}
+	require.Len(t, h.records, numMessages)
+
+	// The earliest messages should have been evicted from cache
+	// So they should go through again without waiting
+	logger.Info(slog.IntValue(0).String())
+	require.Len(t, h.records, numMessages+1)
+
+	// Recent messages should still be debounced
+	logger.Info(slog.IntValue(numMessages - 1).String())
+	require.Len(t, h.records, numMessages+1)
+}
+
+func TestDebouncingHandler_SameMessageDifferentAttrs(t *testing.T) {
+	h := new(testRecorder)
+	d := NewDebouncingHandler(h)
+	logger := log.NewLogger(d)
+
+	// Log message with one set of attributes
+	logger.Info("same message", "key1", "value1", "key2", "value2")
+	require.Len(t, h.records, 1)
+	require.Equal(t, "same message", h.records[0].Message)
+
+	// Same message with different attributes should still be debounced
+	logger.Info("same message", "key3", "value3", "key4", "value4")
+	require.Len(t, h.records, 1)
+
+	// Same message with no attributes should still be debounced
+	logger.Info("same message")
+	require.Len(t, h.records, 1)
+
+	// Same message with partially overlapping attributes should still be debounced
+	logger.Info("same message", "key1", "different_value", "key5", "value5")
+	require.Len(t, h.records, 1)
+
+	// Wait for debounce period
+	time.Sleep(DebounceDuration + 1*time.Millisecond)
+
+	// Now the same message with any attributes should go through
+	logger.Info("same message", "totally", "new", "attrs", "here")
+	require.Len(t, h.records, 2)
+	require.Equal(t, "same message", h.records[1].Message)
+}
+
+func TestDebouncingHandler_TickerWarning(t *testing.T) {
+	h := new(testRecorder)
+	d := NewDebouncingHandler(h)
+	logger := log.NewLogger(d)
+
+	// Send initial message
+	logger.Info("test message 1")
+	require.Len(t, h.records, 1)
+	require.Equal(t, "test message 1", h.records[0].Message)
+
+	// Trigger several debounced messages
+	for i := 0; i < 10; i++ {
+		logger.Info("test message 1")
+	}
+	// Still only the first message
+	require.Len(t, h.records, 1)
+
+	// Send another unique message and debounce it
+	logger.Info("test message 2")
+	require.Len(t, h.records, 2)
+	for i := 0; i < 5; i++ {
+		logger.Info("test message 2")
+	}
+
+	// Wait for ticker to fire (5 seconds)
+	time.Sleep(DebounceTickerInterval + 100*time.Millisecond)
+
+	// Send a new message to trigger the ticker check
+	logger.Info("trigger ticker check")
+
+	// Should have: original 2 messages, warning about debounced messages, and the trigger message
+	require.Len(t, h.records, 4)
+	require.Equal(t, "test message 1", h.records[0].Message)
+	require.Equal(t, "test message 2", h.records[1].Message)
+	require.Equal(t, DebounceWarningMessage, h.records[2].Message)
+	require.Equal(t, "trigger ticker check", h.records[3].Message)
+
+	// Check that the warning record has the debounced count
+	warningRecord := h.records[2]
+	hasDebounceCount := false
+	warningRecord.Attrs(func(attr slog.Attr) bool {
+		if attr.Key == "nDebounced" {
+			require.Equal(t, uint64(15), attr.Value.Uint64()) // 10 + 5 debounced messages
+			hasDebounceCount = true
+		}
+		return true
+	})
+	require.True(t, hasDebounceCount, "Warning should contain nDebounced attribute")
+
+	// Counter should be reset, so debouncing more messages starts fresh
+	for i := 0; i < 3; i++ {
+		logger.Info("trigger ticker check")
+	}
+	// No new messages should be logged (they're debounced)
+	require.Len(t, h.records, 4)
+
+	// Wait for ticker again
+	time.Sleep(DebounceTickerInterval + 100*time.Millisecond)
+
+	// Trigger ticker check
+	logger.Info("final message")
+
+	// Should have another warning for the 3 newly debounced messages
+	require.Len(t, h.records, 6)
+	require.Equal(t, DebounceWarningMessage, h.records[4].Message)
+	require.Equal(t, "final message", h.records[5].Message)
+
+	// Check the second warning has count of 3
+	secondWarning := h.records[4]
+	secondWarning.Attrs(func(attr slog.Attr) bool {
+		if attr.Key == "nDebounced" {
+			require.Equal(t, uint64(3), attr.Value.Uint64())
+		}
+		return true
+	})
+}
+
+func TestDebouncingHandler_Concurrent(t *testing.T) {
+	h := new(safeTestRecorder)
+	d := NewDebouncingHandler(h)
+	logger := log.NewLogger(d)
+
+	const numGoroutines = 10
+	var wg sync.WaitGroup
+	wg.Add(numGoroutines)
+
+	for i := range numGoroutines {
+		go func(id int) {
+			defer wg.Done()
+			logger.Info("hello")
+		}(i)
+	}
+
+	wg.Wait()
+
+	require.Equal(t, h.Len(), 1, "Should have debounced duplicate messages")
+}
