master: sync hashicorp raft leadership with seaweedfs/raft leader

When seaweedfs/raft elects a new leader, transfer hashicorp raft leadership
to the same node. This ensures both raft systems always have the same leader,
allowing dual-write to work correctly.

Changes:
- Add SyncHashicorpRaftLeadership() to transfer hashicorp raft leadership
- Call sync on seaweedfs/raft leader change event
- Only the current hashicorp raft leader can initiate transfer

weed/command/master.go

@@ -244,6 +244,11 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 		raftServer.TransportManager.Register(grpcS)
 	} else {
 		protobuf.RegisterRaftServer(grpcS, raftServer)
+		// Also register hashicorp transport for dual-write migration
+		if raftServer.TransportManager != nil {
+			raftServer.TransportManager.Register(grpcS)
+			glog.V(0).Infof("Dual-write: hashicorp raft transport registered")
+		}
 	}
 	reflection.Register(grpcS)
 	glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", version.Version(), *masterOption.ipBind, grpcPort)

weed/server/master_server.go

@@ -209,9 +209,16 @@ func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
 			if ms.Topo.RaftServer.Leader() != "" {
 				glog.V(0).Infof("[%s] %s becomes leader.", ms.Topo.RaftServer.Name(), ms.Topo.RaftServer.Leader())
 				ms.Topo.LastLeaderChangeTime = time.Now()
+				// Sync hashicorp raft leadership to match seaweedfs/raft leader for dual-write
+				raftServer.SyncHashicorpRaftLeadership(ms.Topo.RaftServer.Leader())
 			}
 		})
 		raftServerName = fmt.Sprintf("[%s]", ms.Topo.RaftServer.Name())
+		// Also set HashicorpRaft if available for dual-write migration
+		if raftServer.RaftHashicorp != nil {
+			ms.Topo.HashicorpRaft = raftServer.RaftHashicorp
+			glog.V(0).Infof("Dual-write mode: hashicorp raft enabled alongside seaweedfs/raft")
+		}
 	} else if raftServer.RaftHashicorp != nil {
 		ms.Topo.HashicorpRaft = raftServer.RaftHashicorp
 		raftServerName = ms.Topo.HashicorpRaft.String()

weed/server/raft_server.go

@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"encoding/json"
+	"fmt"
 	"io"
 	"math/rand/v2"
 	"os"
@@ -9,6 +10,7 @@ import (
 	"time"
 
 	transport "github.com/Jille/raft-grpc-transport"
+	boltdb "github.com/hashicorp/raft-boltdb/v2"
 
 	"google.golang.org/grpc"
 
@@ -167,6 +169,13 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) {
 
 	glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader())
 
+	// Also initialize hashicorp raft in parallel for seamless future migration
+	if err := s.initHashicorpRaftForDualWrite(option); err != nil {
+		glog.Warningf("failed to initialize hashicorp raft for dual-write, migration will require manual steps: %v", err)
+	} else {
+		glog.V(0).Infof("hashicorp raft initialized for dual-write migration")
+	}
+
 	return s, nil
 }
 
@@ -197,3 +206,111 @@ func (s *RaftServer) DoJoinCommand() {
 	}
 
 }
+
+// initHashicorpRaftForDualWrite initializes hashicorp raft alongside seaweedfs/raft
+// for seamless future migration. State changes are written to both implementations.
+func (s *RaftServer) initHashicorpRaftForDualWrite(option *RaftServerOption) error {
+	c := hashicorpRaft.DefaultConfig()
+	c.LocalID = hashicorpRaft.ServerID(s.serverAddr)
+	c.HeartbeatTimeout = time.Duration(float64(option.HeartbeatInterval) * (rand.Float64()*0.25 + 1))
+	c.ElectionTimeout = option.ElectionTimeout
+	if c.LeaderLeaseTimeout > c.HeartbeatTimeout {
+		c.LeaderLeaseTimeout = c.HeartbeatTimeout
+	}
+	if glog.V(4) {
+		c.LogLevel = "Debug"
+	} else if glog.V(2) {
+		c.LogLevel = "Info"
+	} else if glog.V(1) {
+		c.LogLevel = "Warn"
+	} else if glog.V(0) {
+		c.LogLevel = "Error"
+	}
+
+	if err := hashicorpRaft.ValidateConfig(c); err != nil {
+		return fmt.Errorf("raft.ValidateConfig: %w", err)
+	}
+
+	// Use a subdirectory to keep hashicorp raft data separate from old raft
+	hashicorpDataDir := path.Join(s.dataDir, "hashicorp")
+	if option.RaftBootstrap {
+		os.RemoveAll(path.Join(hashicorpDataDir, ldbFile))
+		os.RemoveAll(path.Join(hashicorpDataDir, sdbFile))
+		os.RemoveAll(path.Join(hashicorpDataDir, "snapshots"))
+	}
+	if err := os.MkdirAll(path.Join(hashicorpDataDir, "snapshots"), os.ModePerm); err != nil {
+		return err
+	}
+
+	ldb, err := boltdb.NewBoltStore(path.Join(hashicorpDataDir, ldbFile))
+	if err != nil {
+		return fmt.Errorf("boltdb.NewBoltStore(%q): %v", path.Join(hashicorpDataDir, ldbFile), err)
+	}
+
+	sdb, err := boltdb.NewBoltStore(path.Join(hashicorpDataDir, sdbFile))
+	if err != nil {
+		return fmt.Errorf("boltdb.NewBoltStore(%q): %v", path.Join(hashicorpDataDir, sdbFile), err)
+	}
+
+	fss, err := hashicorpRaft.NewFileSnapshotStore(hashicorpDataDir, 3, os.Stderr)
+	if err != nil {
+		return fmt.Errorf("raft.NewFileSnapshotStore(%q, ...): %v", hashicorpDataDir, err)
+	}
+
+	s.TransportManager = transport.New(hashicorpRaft.ServerAddress(s.serverAddr), []grpc.DialOption{option.GrpcDialOption})
+
+	stateMachine := StateMachine{topo: option.Topo}
+	s.RaftHashicorp, err = hashicorpRaft.NewRaft(c, &stateMachine, ldb, sdb, fss, s.TransportManager.Transport())
+	if err != nil {
+		return fmt.Errorf("raft.NewRaft: %w", err)
+	}
+
+	if option.RaftBootstrap || len(s.RaftHashicorp.GetConfiguration().Configuration().Servers) == 0 {
+		cfg := s.AddPeersConfiguration()
+		peerIdx := getPeerIdx(s.serverAddr, s.peers)
+		timeSleep := time.Duration(float64(c.LeaderLeaseTimeout) * (rand.Float64()*0.25 + 1) * float64(peerIdx))
+		glog.V(0).Infof("Dual-write: Bootstrapping hashicorp raft idx: %d sleep: %v cluster: %+v", peerIdx, timeSleep, cfg)
+		time.Sleep(timeSleep)
+		f := s.RaftHashicorp.BootstrapCluster(cfg)
+		if err := f.Error(); err != nil {
+			return fmt.Errorf("raft.Raft.BootstrapCluster: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// SyncHashicorpRaftLeadership transfers hashicorp raft leadership to match seaweedfs/raft leader.
+// This ensures both raft systems have the same leader for dual-write to work correctly.
+func (s *RaftServer) SyncHashicorpRaftLeadership(seaweedfsRaftLeader string) {
+	if s.RaftHashicorp == nil || seaweedfsRaftLeader == "" {
+		return
+	}
+
+	hashicorpLeader, _ := s.RaftHashicorp.LeaderWithID()
+	targetLeader := pb.ServerAddress(seaweedfsRaftLeader)
+
+	// If hashicorp raft already has the same leader, nothing to do
+	if string(hashicorpLeader) == targetLeader.ToGrpcAddress() {
+		glog.V(1).Infof("Dual-write: hashicorp raft leader already matches seaweedfs/raft leader: %s", seaweedfsRaftLeader)
+		return
+	}
+
+	// Only the current hashicorp raft leader can transfer leadership
+	if s.RaftHashicorp.State() != hashicorpRaft.Leader {
+		glog.V(1).Infof("Dual-write: not hashicorp raft leader, cannot transfer leadership to %s", seaweedfsRaftLeader)
+		return
+	}
+
+	// Transfer leadership to the seaweedfs/raft leader
+	glog.V(0).Infof("Dual-write: transferring hashicorp raft leadership to match seaweedfs/raft leader: %s", seaweedfsRaftLeader)
+	future := s.RaftHashicorp.LeadershipTransferToServer(
+		hashicorpRaft.ServerID(targetLeader),
+		hashicorpRaft.ServerAddress(targetLeader.ToGrpcAddress()),
+	)
+	if err := future.Error(); err != nil {
+		glog.Warningf("Dual-write: failed to transfer hashicorp raft leadership to %s: %v", seaweedfsRaftLeader, err)
+	} else {
+		glog.V(0).Infof("Dual-write: hashicorp raft leadership transferred to %s", seaweedfsRaftLeader)
+	}
+}

weed/topology/topology.go

@@ -237,6 +237,15 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
 		if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next)); err != nil {
 			return 0, err
 		}
+		// Dual-write to hashicorp raft if available for seamless migration
+		if t.HashicorpRaft != nil {
+			b, err := json.Marshal(NewMaxVolumeIdCommand(next))
+			if err != nil {
+				glog.Warningf("failed to marshal MaxVolumeIdCommand for dual-write: %v", err)
+			} else if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
+				glog.Warningf("failed to apply MaxVolumeIdCommand to hashicorp raft for dual-write: %v", future.Error())
+			}
+		}
 	} else if t.HashicorpRaft != nil {
 		b, err := json.Marshal(NewMaxVolumeIdCommand(next))
 		if err != nil {