fix(raft): address julienrbrt review — logger, boltdb filter, ShutdownTimeout

- Remove stdlib log filter (boltTxClosedFilter / suppressBoltNoise): it
  redirected the global stdlib logger which is the wrong scope. raft-boltdb
  uses log.Printf directly with no Logger option, so the "tx closed" noise
  is now accepted as-is from stderr.

- Wire hashicorp/raft's internal hclog output through zerolog: set
  raft.Config.Logger to an hclog.Logger backed by the zerolog writer so
  all raft-internal messages appear in the structured log stream under
  component=raft-hashicorp.

- Remove ShutdownTimeout from public config: it was a second "how long to
  wait" knob that confused operators. ShutdownTimeout is now computed
  internally as 5 × SendTimeout at the initRaftNode call site.

- Delete TestRaftRetrieverStopClearsApplyCallback: tested an implementation
  detail (Stop clears the apply callback pointer) rather than observable
  behaviour. The stubRaftNode helper it defined is moved to syncer_test.go
  where it is still needed.

- Rename TestNewNode_SnapshotConfigApplied → TestBuildRaftConfig_SnapshotConfigApplied
  to reflect that it tests buildRaftConfig, not NewNode.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: auricom

block/internal/syncing/raft_retriever_test.go

@@ -6,6 +6,7 @@ import (
 	"crypto/sha512"
 	"errors"
 	"math"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"testing/synctest"
@@ -35,6 +36,30 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
+// stubRaftNode is a minimal RaftNode stub that records SetApplyCallback calls.
+type stubRaftNode struct {
+	mu        sync.Mutex
+	callbacks []chan<- raft.RaftApplyMsg
+}
+
+func (s *stubRaftNode) IsLeader() bool                                          { return false }
+func (s *stubRaftNode) HasQuorum() bool                                         { return false }
+func (s *stubRaftNode) GetState() *raft.RaftBlockState                          { return nil }
+func (s *stubRaftNode) Broadcast(context.Context, *raft.RaftBlockState) error   { return nil }
+func (s *stubRaftNode) SetApplyCallback(ch chan<- raft.RaftApplyMsg) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.callbacks = append(s.callbacks, ch)
+}
+
+func (s *stubRaftNode) recordedCallbacks() []chan<- raft.RaftApplyMsg {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]chan<- raft.RaftApplyMsg, len(s.callbacks))
+	copy(out, s.callbacks)
+	return out
+}
+
 // helper to create a signer, pubkey and address for tests
 func buildSyncTestSigner(tb testing.TB) (addr []byte, pub crypto.PubKey, signer signerpkg.Signer) {
 	tb.Helper()

block/internal/syncing/syncer_test.go

@@ -155,7 +155,7 @@ func initRaftNode(nodeConfig config.Config, logger zerolog.Logger) (*raftpkg.Nod
 		Bootstrap:          nodeConfig.Raft.Bootstrap,
 		SnapCount:          nodeConfig.Raft.SnapCount,
 		SendTimeout:        nodeConfig.Raft.SendTimeout,
-		ShutdownTimeout:    nodeConfig.Raft.ShutdownTimeout,
+		ShutdownTimeout:    5 * nodeConfig.Raft.SendTimeout,
 		HeartbeatTimeout:   nodeConfig.Raft.HeartbeatTimeout,
 		LeaderLeaseTimeout: nodeConfig.Raft.LeaderLeaseTimeout,
 		ElectionTimeout:    nodeConfig.Raft.ElectionTimeout,

node/full.go

@@ -201,9 +201,6 @@ const (
 	FlagRaftSnapshotThreshold = FlagPrefixEvnode + "raft.snapshot_threshold"
 	// FlagRaftTrailingLogs is the flag for the number of trailing logs after a snapshot.
 	FlagRaftTrailingLogs = FlagPrefixEvnode + "raft.trailing_logs"
-	// FlagRaftShutdownTimeout is the flag for how long to wait for committed logs to be applied on graceful shutdown.
-	FlagRaftShutdownTimeout = FlagPrefixEvnode + "raft.shutdown_timeout"
-
 	// Pruning configuration flags
 	FlagPruningMode       = FlagPrefixEvnode + "pruning.pruning_mode"
 	FlagPruningKeepRecent = FlagPrefixEvnode + "pruning.pruning_keep_recent"
@@ -412,7 +409,6 @@ type RaftConfig struct {
 	Peers              string        `mapstructure:"peers" yaml:"peers" comment:"Comma-separated list of peer Raft addresses (nodeID@host:port)"`
 	SnapCount          uint64        `mapstructure:"snap_count" yaml:"snap_count" comment:"Number of snapshot files to retain on disk"`
 	SendTimeout        time.Duration `mapstructure:"send_timeout" yaml:"send_timeout" comment:"Max duration to wait for a message to be sent to a peer"`
-	ShutdownTimeout    time.Duration `mapstructure:"shutdown_timeout" yaml:"shutdown_timeout" comment:"Max duration to wait for committed raft logs to be applied on graceful shutdown"`
 	HeartbeatTimeout   time.Duration `mapstructure:"heartbeat_timeout" yaml:"heartbeat_timeout" comment:"Time between leader heartbeats to followers"`
 	LeaderLeaseTimeout time.Duration `mapstructure:"leader_lease_timeout" yaml:"leader_lease_timeout" comment:"Duration of the leader lease"`
 	ElectionTimeout    time.Duration `mapstructure:"election_timeout" yaml:"election_timeout" comment:"Time a candidate waits for votes before restarting election; must be >= heartbeat_timeout"`
@@ -439,10 +435,6 @@ func (c RaftConfig) Validate() error {
 		multiErr = errors.Join(multiErr, fmt.Errorf("send timeout must be positive"))
 	}
 
-	if c.ShutdownTimeout <= 0 {
-		multiErr = errors.Join(multiErr, fmt.Errorf("shutdown timeout must be positive"))
-	}
-
 	if c.HeartbeatTimeout <= 0 {
 		multiErr = errors.Join(multiErr, fmt.Errorf("heartbeat timeout must be positive"))
 	}
@@ -672,7 +664,6 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().String(FlagRaftPeers, def.Raft.Peers, "comma-separated list of peer Raft addresses (nodeID@host:port)")
 	cmd.Flags().Uint64(FlagRaftSnapCount, def.Raft.SnapCount, "number of snapshot files to retain on disk")
 	cmd.Flags().Duration(FlagRaftSendTimeout, def.Raft.SendTimeout, "max duration to wait for a message to be sent to a peer")
-	cmd.Flags().Duration(FlagRaftShutdownTimeout, def.Raft.ShutdownTimeout, "max duration to wait for committed raft logs to be applied on graceful shutdown")
 	cmd.Flags().Duration(FlagRaftHeartbeatTimeout, def.Raft.HeartbeatTimeout, "time between leader heartbeats to followers")
 	cmd.Flags().Duration(FlagRaftLeaderLeaseTimeout, def.Raft.LeaderLeaseTimeout, "duration of the leader lease")
 	cmd.Flags().Duration(FlagRaftElectionTimeout, def.Raft.ElectionTimeout, "time a candidate waits for votes before restarting election")

pkg/config/config.go

@@ -131,7 +131,6 @@ func TestAddFlags(t *testing.T) {
 	assertFlagValue(t, flags, FlagRaftPeers, DefaultConfig().Raft.Peers)
 	assertFlagValue(t, flags, FlagRaftSnapCount, DefaultConfig().Raft.SnapCount)
 	assertFlagValue(t, flags, FlagRaftSendTimeout, DefaultConfig().Raft.SendTimeout)
-	assertFlagValue(t, flags, FlagRaftShutdownTimeout, DefaultConfig().Raft.ShutdownTimeout)
 	assertFlagValue(t, flags, FlagRaftHeartbeatTimeout, DefaultConfig().Raft.HeartbeatTimeout)
 	assertFlagValue(t, flags, FlagRaftLeaderLeaseTimeout, DefaultConfig().Raft.LeaderLeaseTimeout)
 	assertFlagValue(t, flags, FlagRaftElectionTimeout, DefaultConfig().Raft.ElectionTimeout)
@@ -144,7 +143,7 @@ func TestAddFlags(t *testing.T) {
 	assertFlagValue(t, flags, FlagPruningInterval, DefaultConfig().Pruning.Interval.Duration)
 
 	// Count the number of flags we're explicitly checking
-	expectedFlagCount := 82 // Update this number if you add more flag checks above
+	expectedFlagCount := 81 // Update this number if you add more flag checks above
 
 	// Get the actual number of flags (both regular and persistent)
 	actualFlagCount := 0
@@ -405,7 +404,6 @@ func TestRaftConfig_Validate(t *testing.T) {
 			Peers:              "",
 			SnapCount:          1,
 			SendTimeout:        1 * time.Second,
-			ShutdownTimeout:    5 * time.Second,
 			HeartbeatTimeout:   1 * time.Second,
 			LeaderLeaseTimeout: 1 * time.Second,
 			ElectionTimeout:    2 * time.Second,
@@ -446,10 +444,6 @@ func TestRaftConfig_Validate(t *testing.T) {
 			mutate: func(c *RaftConfig) { c.LeaderLeaseTimeout = 0 },
 			expErr: "leader lease timeout must be positive",
 		},
-		"non-positive shutdown timeout": {
-			mutate: func(c *RaftConfig) { c.ShutdownTimeout = 0 },
-			expErr: "shutdown timeout must be positive",
-		},
 		"negative election timeout rejected": {
 			mutate: func(c *RaftConfig) { c.ElectionTimeout = -1 * time.Second },
 			expErr: "election timeout (-1s) must be >= 0",

pkg/config/config_test.go

@@ -116,7 +116,6 @@ func DefaultConfig() Config {
 		},
 		Raft: RaftConfig{
 			SendTimeout:        200 * time.Millisecond,
-			ShutdownTimeout:    5 * time.Second,
 			HeartbeatTimeout:   350 * time.Millisecond,
 			LeaderLeaseTimeout: 175 * time.Millisecond,
 			ElectionTimeout:    1000 * time.Millisecond,

pkg/config/defaults.go

@@ -1,12 +1,10 @@
 package raft
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io"
-	"log"
 	"net"
 	"os"
 	"path/filepath"
@@ -15,30 +13,13 @@ import (
 	"sync/atomic"
 	"time"
 
+	hclog "github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/raft"
 	raftboltdb "github.com/hashicorp/raft-boltdb"
 	"github.com/rs/zerolog"
 	"google.golang.org/protobuf/proto"
 )
 
-// suppressBoltNoise redirects the stdlib log output once to drop the
-// "Rollback failed: tx closed" messages emitted by hashicorp/raft-boltdb.
-// boltdb returns ErrTxClosed when Rollback is called after a successful
-// Commit; raft-boltdb unconditionally logs this as an error even though it
-// is the expected outcome of every successful transaction.
-var suppressBoltNoise sync.Once
-
-// boltTxClosedFilter is an io.Writer that silently drops log lines containing
-// "tx closed" and forwards everything else to the underlying writer.
-type boltTxClosedFilter struct{ w io.Writer }
-
-func (f *boltTxClosedFilter) Write(p []byte) (n int, err error) {
-	if bytes.Contains(p, []byte("tx closed")) {
-		return len(p), nil
-	}
-	return f.w.Write(p)
-}
-
 // Node represents a raft consensus node
 type Node struct {
 	raft   *raft.Raft
@@ -73,10 +54,19 @@ type FSM struct {
 }
 
 // buildRaftConfig converts a Node Config into a hashicorp/raft Config.
-func buildRaftConfig(cfg *Config) *raft.Config {
+// logger is used to bridge hashicorp/raft's internal hclog output to zerolog.
+func buildRaftConfig(cfg *Config, logger zerolog.Logger) *raft.Config {
 	raftConfig := raft.DefaultConfig()
 	raftConfig.LocalID = raft.ServerID(cfg.NodeID)
-	raftConfig.LogLevel = "INFO"
+	// Route raft's internal hclog messages through zerolog so all log output is
+	// consistent. hclog writes formatted text lines; zerolog receives them via
+	// its io.Writer implementation and emits them as structured JSON.
+	raftConfig.Logger = hclog.New(&hclog.LoggerOptions{
+		Name:        "raft",
+		Level:       hclog.Info,
+		Output:      logger.With().Str("component", "raft-hashicorp").Logger(),
+		DisableTime: true, // zerolog adds its own timestamp
+	})
 	raftConfig.HeartbeatTimeout = cfg.HeartbeatTimeout
 	raftConfig.LeaderLeaseTimeout = cfg.LeaderLeaseTimeout
 	if cfg.ElectionTimeout > 0 {
@@ -92,14 +82,11 @@ func buildRaftConfig(cfg *Config) *raft.Config {
 }
 
 func NewNode(cfg *Config, logger zerolog.Logger) (*Node, error) {
-	suppressBoltNoise.Do(func() {
-		log.SetOutput(&boltTxClosedFilter{w: log.Writer()})
-	})
 	if err := os.MkdirAll(cfg.RaftDir, 0755); err != nil {
 		return nil, fmt.Errorf("create raft dir: %w", err)
 	}
 
-	raftConfig := buildRaftConfig(cfg)
+	raftConfig := buildRaftConfig(cfg, logger)
 
 	startPointer := new(atomic.Pointer[RaftBlockState])
 	startPointer.Store(&RaftBlockState{})

pkg/raft/node.go

@@ -1,52 +1,17 @@
 package raft
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"testing"
 	"time"
 
 	"github.com/hashicorp/raft"
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-func TestBoltTxClosedFilter_Write(t *testing.T) {
-	specs := map[string]struct {
-		input     string
-		expectFwd bool
-	}{
-		"passes through normal log line": {
-			input:     "some normal log message\n",
-			expectFwd: true,
-		},
-		"drops line containing tx closed": {
-			input:     "Rollback failed: tx closed\n",
-			expectFwd: false,
-		},
-		"drops line with tx closed anywhere": {
-			input:     "error: tx closed due to commit\n",
-			expectFwd: false,
-		},
-	}
-
-	for name, spec := range specs {
-		t.Run(name, func(t *testing.T) {
-			var buf bytes.Buffer
-			f := &boltTxClosedFilter{w: &buf}
-			n, err := f.Write([]byte(spec.input))
-			require.NoError(t, err)
-			assert.Equal(t, len(spec.input), n)
-			if spec.expectFwd {
-				assert.Equal(t, spec.input, buf.String())
-			} else {
-				assert.Empty(t, buf.String())
-			}
-		})
-	}
-}
-
 func TestBuildRaftConfig_ElectionTimeout(t *testing.T) {
 	specs := map[string]struct {
 		cfg                     *Config
@@ -70,7 +35,7 @@ func TestBuildRaftConfig_ElectionTimeout(t *testing.T) {
 
 	for name, spec := range specs {
 		t.Run(name, func(t *testing.T) {
-			rc := buildRaftConfig(spec.cfg)
+			rc := buildRaftConfig(spec.cfg, zerolog.Nop())
 			assert.Equal(t, spec.expectedElectionTimeout, rc.ElectionTimeout)
 		})
 	}
@@ -192,7 +157,7 @@ func TestNodeResignLeader_NotLeaderNoop(t *testing.T) {
 	assert.NoError(t, n.ResignLeader(context.Background()))
 }
 
-func TestNewNode_SnapshotConfigApplied(t *testing.T) {
+func TestBuildRaftConfig_SnapshotConfigApplied(t *testing.T) {
 	specs := map[string]struct {
 		cfg                       *Config
 		expectedSnapshotThreshold uint64
@@ -220,7 +185,7 @@ func TestNewNode_SnapshotConfigApplied(t *testing.T) {
 
 	for name, spec := range specs {
 		t.Run(name, func(t *testing.T) {
-			rc := buildRaftConfig(spec.cfg)
+			rc := buildRaftConfig(spec.cfg, zerolog.Nop())
 			assert.Equal(t, spec.expectedSnapshotThreshold, rc.SnapshotThreshold)
 			assert.Equal(t, spec.expectedTrailingLogs, rc.TrailingLogs)
 		})