chore(autofmt): Automated formatting

go.mod

@@ -45,7 +45,6 @@ require (
 	github.com/jpillora/backoff v1.0.0
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
 	github.com/lni/dragonboat/v4 v4.0.0-20240618143154-6a1623140f27
-	github.com/lni/goutils v1.4.0
 	github.com/mattn/go-isatty v0.0.20
 	github.com/multiformats/go-base36 v0.2.0
 	github.com/opencontainers/go-digest v1.0.0
@@ -159,6 +158,7 @@ require (
 	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/lni/goutils v1.4.0 // indirect
 	github.com/lni/vfs v0.2.1-0.20220616104132-8852fd867376 // indirect
 	github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect

internal/raft/cluster.go

@@ -20,10 +20,11 @@ type RaftConfig struct {
 	RaftAddress    string   `help:"Address to advertise to other nodes" required:""`
 	ListenAddress  string   `help:"Address to listen for incoming traffic. If empty, RaftAddress will be used."`
 	// Raft configuration
-	ElectionRTT        uint64 `help:"Election RTT" default:"10"`
-	HeartbeatRTT       uint64 `help:"Heartbeat RTT" default:"1"`
-	SnapshotEntries    uint64 `help:"Snapshot entries" default:"10"`
-	CompactionOverhead uint64 `help:"Compaction overhead" default:"100"`
+	RTT                time.Duration `help:"Estimated average round trip time between nodes" default:"200ms"`
+	ElectionRTT        uint64        `help:"Election RTT as a multiple of RTT" default:"10"`
+	HeartbeatRTT       uint64        `help:"Heartbeat RTT as a multiple of RTT" default:"1"`
+	SnapshotEntries    uint64        `help:"Snapshot entries" default:"10"`
+	CompactionOverhead uint64        `help:"Compaction overhead" default:"100"`
 }
 
 // Cluster of dragonboat nodes.
@@ -96,12 +97,16 @@ func New(cfg *RaftConfig) *Cluster {
 }
 
 // AddShard adds a shard to the cluster.
-func AddShard[Q any, R any, E Event, EPtr Unmasrshallable[E]](
+// This can be only called before the cluster is started.
+func AddShard[Q any, R any, E Event, EPtr Unmarshallable[E]](
 	ctx context.Context,
 	to *Cluster,
 	shardID uint64,
 	sm StateMachine[Q, R, E, EPtr],
 ) *ShardHandle[E, Q, R] {
+	if to.nh != nil {
+		panic("cluster already started")
+	}
 	to.shards[shardID] = newStateMachineShim[Q, R, E, EPtr](sm)
 	return &ShardHandle[E, Q, R]{
 		shardID: shardID,
@@ -124,7 +129,7 @@ func (c *Cluster) start(ctx context.Context, join bool) error {
 	nhc := config.NodeHostConfig{
 		WALDir:         c.config.DataDir,
 		NodeHostDir:    c.config.DataDir,
-		RTTMillisecond: 200,
+		RTTMillisecond: uint64(c.config.RTT.Milliseconds()),
 		RaftAddress:    c.config.RaftAddress,
 		ListenAddress:  c.config.ListenAddress,
 	}

internal/raft/cluster_test.go

@@ -40,7 +40,7 @@ func (s *IntStateMachine) Save(writer io.Writer) error     { return nil }
 func (s *IntStateMachine) Close() error                    { return nil }
 
 func TestCluster(t *testing.T) {
-	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(30*time.Second))
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(60*time.Second))
 	defer cancel()
 
 	members := []string{"localhost:51001", "localhost:51002"}
@@ -71,7 +71,7 @@ func TestCluster(t *testing.T) {
 }
 
 func TestJoiningExistingCluster(t *testing.T) {
-	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(30*time.Second))
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(60*time.Second))
 	defer cancel()
 
 	members := []string{"localhost:51001", "localhost:51002"}

internal/raft/statemachine.go

@@ -13,8 +13,8 @@ type Event interface {
 	encoding.BinaryMarshaler
 }
 
-// Unmasrshallable is a type that can be unmarshalled from a binary representation.
-type Unmasrshallable[T any] interface {
+// Unmarshallable is a type that can be unmarshalled from a binary representation.
+type Unmarshallable[T any] interface {
 	*T
 	encoding.BinaryUnmarshaler
 }
@@ -25,7 +25,7 @@ type Unmasrshallable[T any] interface {
 // Q is the query type.
 // R is the query response type.
 // E is the event type.
-type StateMachine[Q any, R any, E Event, EPtr Unmasrshallable[E]] interface {
+type StateMachine[Q any, R any, E Event, EPtr Unmarshallable[E]] interface {
 	// Query the state of the state machine.
 	Lookup(key Q) (R, error)
 	// Update the state of the state machine.
@@ -38,11 +38,11 @@ type StateMachine[Q any, R any, E Event, EPtr Unmasrshallable[E]] interface {
 	Close() error
 }
 
-type stateMachineShim[Q any, R any, E Event, EPtr Unmasrshallable[E]] struct {
+type stateMachineShim[Q any, R any, E Event, EPtr Unmarshallable[E]] struct {
 	sm StateMachine[Q, R, E, EPtr]
 }
 
-func newStateMachineShim[Q any, R any, E Event, EPtr Unmasrshallable[E]](
+func newStateMachineShim[Q any, R any, E Event, EPtr Unmarshallable[E]](
 	sm StateMachine[Q, R, E, EPtr],
 ) statemachine.CreateStateMachineFunc {
 	return func(clusterID uint64, nodeID uint64) statemachine.IStateMachine {