From 1eea1a9f3593a3a7947c3486d552db39e4a88e48 Mon Sep 17 00:00:00 2001
From: Mariem Baccari <mariem.baccari@orbitalize.com>
Date: Wed, 20 May 2026 11:12:05 +0200
Subject: [PATCH 1/5] implement consensus logic

---
 pkg/raftstore/consensus/consensus.go | 174 +++++++++++++++++++++++++--
 pkg/raftstore/params/params.go       |  71 +++++++++--
 pkg/raftstore/store.go               |   2 +-
 3 files changed, 227 insertions(+), 20 deletions(-)

diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go
index 2541b9141..517fb3a0e 100644
--- a/pkg/raftstore/consensus/consensus.go
+++ b/pkg/raftstore/consensus/consensus.go
@@ -6,8 +6,10 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/interuss/dss/pkg/logging"
+	params "github.com/interuss/dss/pkg/raftstore/params"
 	"github.com/interuss/stacktrace"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
@@ -17,43 +19,99 @@ import (
 	"go.uber.org/zap"
 )
 
-const (
-	defaultClusterID uint64 = 1
-)
+const serverShutdownTimeout = 5 * time.Second
 
 type Consensus struct {
 	logger *zap.Logger
 
-	node raft.Node
+	nodeID uint64
+	node   raft.Node
 
 	transport *rafthttp.Transport
 	server    *http.Server
 
 	storage *storage
-	errorC  chan error
+
+	confState     raftpb.ConfState
+	snapshotIndex uint64
+	appliedIndex  uint64
 }
 
-func NewConsensus(ctx context.Context, logger *zap.Logger, nodeID uint64, peers map[uint64]*url.URL, dataDir string, snapshotCatchupEntries uint64) (*Consensus, error) {
-	storage, _, err := newStorage(ctx, logger.With(zap.String("component", "storage")), dataDir, nodeID, snapshotCatchupEntries)
+func NewConsensus(ctx context.Context, logger *zap.Logger, peers map[uint64]*url.URL, connectParams params.ConnectParameters) (*Consensus, error) {
+	storage, old, err := newStorage(ctx, logger.With(zap.String("component", "storage")), connectParams.DataDir, connectParams.NodeID, connectParams.SnapshotCatchupEntries)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to initialize storage")
 	}
 
+	nodeUrl, ok := peers[connectParams.NodeID]
+	if !ok {
+		return nil, stacktrace.NewError("node ID %d not found in peers map", connectParams.NodeID)
+	}
+
+	var node raft.Node
+	config := connectParams.RaftConfig(storage)
+	if old {
+		logger.Info("restarting raft node", zap.String("address", nodeUrl.String()))
+		node = raft.RestartNode(config)
+	} else {
+		logger.Info("starting new raft node", zap.String("address", nodeUrl.String()))
+		node = raft.StartNode(config, peersList(peers))
+	}
+
 	consensus := &Consensus{
 		logger: logging.WithValuesFromContext(ctx, logger),
 
+		nodeID: connectParams.NodeID,
+		node:   node,
+
 		storage: storage,
-		errorC:  make(chan error, 1),
 	}
 
-	err = consensus.initTransport(ctx, nodeID, defaultClusterID, peers)
+	err = consensus.initTransport(ctx, connectParams.NodeID, connectParams.ClusterID, peers)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to initialize transport")
 	}
 
+	snap, err := consensus.storage.Snapshot()
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to get snapshot from storage")
+	}
+
+	consensus.confState = snap.Metadata.ConfState
+	consensus.snapshotIndex = snap.Metadata.Index
+	consensus.appliedIndex = snap.Metadata.Index
+
+	go func() {
+		err := consensus.handleReady(connectParams.TickInterval)
+		if err != nil {
+			consensus.logger.Error("handleReady exited with error, shutting down consensus", zap.Error(err))
+		}
+
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), serverShutdownTimeout)
+		defer cancel()
+		if shutdownErr := consensus.server.Shutdown(shutdownCtx); shutdownErr != nil {
+			consensus.logger.Error("failed to shutdown http server", zap.Error(shutdownErr))
+		} else {
+			consensus.logger.Info("http server shutdown complete")
+		}
+
+		consensus.transport.Stop()
+		consensus.logger.Info("transport stopped")
+		consensus.node.Stop()
+		consensus.logger.Info("raft node stopped")
+	}()
+
 	return consensus, nil
 }
 
+func peersList(peers map[uint64]*url.URL) []raft.Peer {
+	result := make([]raft.Peer, 0, len(peers))
+	for id := range peers {
+		result = append(result, raft.Peer{ID: id})
+	}
+	return result
+}
+
 func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL) error {
 	nodeIDStr := fmt.Sprintf("%d", nodeID)
 
@@ -64,7 +122,7 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		Raft:        c,
 		ServerStats: v2stats.NewServerStats(nodeIDStr, nodeIDStr),
 		LeaderStats: v2stats.NewLeaderStats(c.logger, nodeIDStr),
-		ErrorC:      c.errorC,
+		ErrorC:      make(chan error),
 	}
 
 	err := transport.Start()
@@ -95,7 +153,7 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		err := c.server.ListenAndServe()
 		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			c.logger.Error("http server error", zap.Error(err))
-			c.errorC <- err
+			c.transport.ErrorC <- err
 		}
 	}()
 
@@ -103,6 +161,100 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 	return nil
 }
 
+// handleReady processes the Ready channel of the Raft node and applies committed entries to the state machine
+func (c *Consensus) handleReady(tickInterval time.Duration) error {
+	ticker := time.NewTicker(tickInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			c.node.Tick()
+		case err := <-c.transport.ErrorC:
+			return stacktrace.Propagate(err, "transport error")
+		case rd, ok := <-c.node.Ready():
+			if !ok {
+				return stacktrace.NewError("could not read from Ready(), shutting down handler")
+			}
+
+			err := c.storage.handleReceivedState(rd.Snapshot, rd.HardState, rd.Entries)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to handle received snapshot")
+			}
+
+			if !raft.IsEmptySnap(rd.Snapshot) {
+				if rd.Snapshot.Metadata.Index <= c.appliedIndex {
+					return stacktrace.NewError("snapshot index %d shall be greater than current applied index %d", rd.Snapshot.Metadata.Index, c.appliedIndex)
+				}
+
+				err = c.dispatchSnapshot(rd.Snapshot.Data)
+				if err != nil {
+					return stacktrace.Propagate(err, "failed to dispatch snapshot")
+				}
+
+				c.confState = rd.Snapshot.Metadata.ConfState
+				c.snapshotIndex = rd.Snapshot.Metadata.Index
+				c.appliedIndex = rd.Snapshot.Metadata.Index
+			}
+
+			c.updateSnapshotConfState(rd.Messages)
+			c.transport.Send(rd.Messages)
+
+			entries, err := c.entriesToApply(rd.CommittedEntries)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to get entries to apply")
+			}
+
+			err = c.publishEntries(entries)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to publish entries")
+			}
+
+			c.node.Advance()
+		}
+	}
+}
+
+// TODO implement
+func (c *Consensus) publishEntries(_ []raftpb.Entry) error {
+	return nil
+}
+
+// TODO implement
+func (c *Consensus) dispatchSnapshot(_ []byte) error {
+	return nil
+}
+
+func (c *Consensus) entriesToApply(entries []raftpb.Entry) ([]raftpb.Entry, error) {
+	if len(entries) == 0 {
+		return entries, nil
+	}
+
+	result := make([]raftpb.Entry, 0)
+
+	firstIdx := entries[0].Index
+	if firstIdx > c.appliedIndex+1 {
+		return nil, stacktrace.NewError("unexpected gap: first committed entry index %d > applied index %d + 1", firstIdx, c.appliedIndex)
+	}
+
+	// Skip entries that have already been applied.
+	if skip := c.appliedIndex + 1 - firstIdx; skip < uint64(len(entries)) {
+		result = entries[skip:]
+	}
+
+	return result, nil
+}
+
+// updateSnapshotConfState updates the ConfState in the snapshot
+// of messages that contain one as it could be outdated.
+func (c *Consensus) updateSnapshotConfState(msgs []raftpb.Message) {
+	for i := range msgs {
+		if msgs[i].Type == raftpb.MsgSnap {
+			msgs[i].Snapshot.Metadata.ConfState = c.confState
+		}
+	}
+}
+
 // Process implements the rafthttp.Raft interface.
 func (c *Consensus) Process(ctx context.Context, m raftpb.Message) error {
 	return c.node.Step(ctx, m)
diff --git a/pkg/raftstore/params/params.go b/pkg/raftstore/params/params.go
index a135813d2..460dcb7b1 100644
--- a/pkg/raftstore/params/params.go
+++ b/pkg/raftstore/params/params.go
@@ -5,8 +5,10 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/interuss/stacktrace"
+	"go.etcd.io/raft/v3"
 )
 
 const (
@@ -14,13 +16,26 @@ const (
 
 	// the default Raft related parameters are the same as the default values used by etcd for the moment.
 	// TODO - review and adjust these parameters as needed based on testing and performance tuning.
-	defaultSnapshotCatchupEntries = 10000
+
+	defaultSnapshotCatchupEntries  = 5000
+	defaultSnapshotIntervalEntries = 10000
+	defaultTickInterval            = 100 * time.Millisecond
+
+	// follower waits 10 x defaultTickInterval without a heartbeat before starting an election
+	defaultElectionTick = 10
+	// leader sends a heartbeat every tick, must be < defaultElectionTick
+	defaultHeartbeatTick = 1
+
+	defaultMaxSizePerMsg   = 1024 * 1024
+	defaultMaxInflightMsgs = 4096 / 8
 )
 
 type (
 	// ConnectParameters bundles up parameters used for connecting nodes in a raftstore cluster.
 	ConnectParameters struct {
-		ID    uint64
+		// unique node identifier within the cluster, 0 is invalid
+		NodeID uint64
+		// comma-separated "nodeID=peerURL" pairs defining all cluster members including this node
 		Peers string
 
 		// DataDir is the directory where the node persists its Raft state (WAL segments and snapshots).
@@ -29,10 +44,26 @@ type (
 		// across restarts unless the node is being permanently shut down.
 		// If the directory is lost, the node will recover by receiving a snapshot from the leader.
 		DataDir string
+		// discriminates this cluster from others sharing the same network, must be identical on all nodes
+		ClusterID uint64
 
-		// SnapshotCatchupEntries is the number of entries for a slow follower to catch-up after compacting.
+		// number of entries for a slow follower to catch-up after compacting.
 		// This gives the follower a buffer of entries while avoiding the need to send a full snapshot.
 		SnapshotCatchupEntries uint64
+		// number of entries applied before triggering a snapshot
+		SnapshotIntervalEntries uint64
+		// base time unit for Raft's logical clock, scales both election and heartbeat timers
+		TickInterval time.Duration
+
+		// ticks without a heartbeat before a follower promotes to candidate, effective timeout = ElectionTick × TickInterval
+		ElectionTick int
+		// ticks between leader heartbeats, must be < ElectionTick
+		HeartbeatTick int
+
+		// max byte size of a message sent to a peer
+		MaxSizePerMsg uint64
+		// max number of in-flight messages during optimistic replication phase
+		MaxInflightMsgs int
 	}
 )
 
@@ -70,16 +101,40 @@ func (c ConnectParameters) PeerMap() (map[uint64]*url.URL, error) {
 	return peers, nil
 }
 
+func (c ConnectParameters) RaftConfig(storage raft.Storage) *raft.Config {
+	return &raft.Config{
+		ID:              c.NodeID,
+		ElectionTick:    c.ElectionTick,
+		HeartbeatTick:   c.HeartbeatTick,
+		MaxSizePerMsg:   c.MaxSizePerMsg,
+		MaxInflightMsgs: c.MaxInflightMsgs,
+		Storage:         storage,
+	}
+}
+
 var (
 	connectParameters ConnectParameters
 )
 
 func init() {
-	flag.Uint64Var(&connectParameters.ID, "raft_node_id", 0, "raft node ID for this instance (must be non-zero and unique within the cluster)")
-	flag.StringVar(&connectParameters.Peers, "raft_peers", "", `comma-separated "nodeID=peerURL" pairs for all cluster members, including the current node, e.g. "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021"`)
-	flag.StringVar(&connectParameters.DataDir, "raft_datadir", defaultDataDir, "directory for raft data (WAL segments and snapshots), required for restarts. These should not be deleted while the node is running or across restarts unless the node is being permanently shut down.")
-
-	flag.Uint64Var(&connectParameters.SnapshotCatchupEntries, "raft_snapshot_catchup_entries", defaultSnapshotCatchupEntries, "number of entries for a slow follower to catch-up after compacting")
+	flag.Uint64Var(&connectParameters.NodeID, "raft_node_id", 0, "Raft node ID for this instance (must be non-zero and unique within the cluster).")
+	flag.Uint64Var(&connectParameters.ClusterID, "raft_cluster_id", 1, "ID of the cluster, used to isolate different Raft clusters running in the same network (must be the same for all nodes in the cluster).")
+	flag.StringVar(&connectParameters.Peers, "raft_peers", "", `Comma-separated "nodeID=peerURL" pairs for all cluster members, including the current node, e.g. "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021"`)
+	flag.StringVar(&connectParameters.DataDir, "raft_datadir", defaultDataDir, "Directory for raft data (WAL segments and snapshots), required for restarts. These should not be deleted while the node is running or across restarts unless the node is being permanently shut down.")
+
+	flag.Uint64Var(&connectParameters.SnapshotCatchupEntries, "raft_snapshot_catchup_entries", defaultSnapshotCatchupEntries,
+		"Log entries retained after compaction so a slow follower can catch up via replication rather than a full snapshot. Higher values tolerate slower followers but increase disk usage.")
+	flag.Uint64Var(&connectParameters.SnapshotIntervalEntries, "raft_snapshot_interval_entries", defaultSnapshotIntervalEntries,
+		"Applied Raft log entries to accumulate before triggering a snapshot. Lower values reduce recovery time but increase I/O frequency.")
+	flag.DurationVar(&connectParameters.TickInterval, "raft_tick_interval", defaultTickInterval,
+		"Base time unit for Raft's logical clock. Election timeout = raft_election_tick x this value; heartbeat interval = raft_heartbeat_tick x this value. Smaller values improve responsiveness but increase network traffic.")
+
+	flag.IntVar(&connectParameters.ElectionTick, "raft_election_tick", defaultElectionTick,
+		"Ticks a follower waits without a leader heartbeat before starting an election. Effective timeout = raft_election_tick x raft_tick_interval. Must be greater than raft_heartbeat_tick. Higher values tolerate slow leaders but delay failover.")
+	flag.IntVar(&connectParameters.HeartbeatTick, "raft_heartbeat_tick", defaultHeartbeatTick,
+		"Ticks between leader heartbeats. Effective interval = raft_heartbeat_tick x raft_tick_interval. Must be less than raft_election_tick. Lower values detect follower loss faster but increase network traffic.")
+	flag.Uint64Var(&connectParameters.MaxSizePerMsg, "raft_max_size_per_msg", defaultMaxSizePerMsg, "Maximum bytes in a single Raft message sent to a peer. Smaller values lower the recovery cost but increase the number of messages sent, affecting throughput during replication.")
+	flag.IntVar(&connectParameters.MaxInflightMsgs, "raft_max_inflight_msgs", defaultMaxInflightMsgs, "Maximum number of in-flight Raft messages during optimistic replication phase. This should be set to avoid overflowing the transport layer sending buffer.")
 }
 
 // GetConnectParameters returns a ConnectParameters instance that gets populated from well-known CLI flags.
diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go
index b602d31c3..3ef62c157 100644
--- a/pkg/raftstore/store.go
+++ b/pkg/raftstore/store.go
@@ -31,7 +31,7 @@ func Init[R any](ctx context.Context, logger *zap.Logger, newRepo func() R) (*St
 			return
 		}
 
-		sharedConsensus, sharedConsensusErr = consensus.NewConsensus(ctx, logger, params.ID, peers, params.DataDir, params.SnapshotCatchupEntries)
+		sharedConsensus, sharedConsensusErr = consensus.NewConsensus(ctx, logger, peers, params)
 		if sharedConsensusErr != nil {
 			sharedConsensusErr = stacktrace.Propagate(sharedConsensusErr, "failed to initialize consensus")
 		}

From 4e1735ccb5bddbe8c310bc5593e11f3ae6d7191b Mon Sep 17 00:00:00 2001
From: Mariem Baccari <mariem.baccari@orbitalize.com>
Date: Wed, 20 May 2026 16:38:42 +0200
Subject: [PATCH 2/5] implement consensus logic part 2

---
 pkg/raftstore/consensus/consensus.go | 176 +++++++++++++++++++++++++--
 pkg/raftstore/consensus/proposal.go  |  65 ++++++++++
 pkg/raftstore/store.go               |   2 +-
 3 files changed, 232 insertions(+), 11 deletions(-)
 create mode 100644 pkg/raftstore/consensus/proposal.go

diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go
index 517fb3a0e..cca7e53ba 100644
--- a/pkg/raftstore/consensus/consensus.go
+++ b/pkg/raftstore/consensus/consensus.go
@@ -2,10 +2,12 @@ package consensus
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
+	"sync"
 	"time"
 
 	"github.com/interuss/dss/pkg/logging"
@@ -17,6 +19,7 @@ import (
 	"go.etcd.io/raft/v3"
 	"go.etcd.io/raft/v3/raftpb"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 )
 
 const serverShutdownTimeout = 5 * time.Second
@@ -30,7 +33,10 @@ type Consensus struct {
 	transport *rafthttp.Transport
 	server    *http.Server
 
-	storage *storage
+	storage   *storage
+	commitChs map[string]chan EntryCommit
+
+	tracker *proposalsTracker
 
 	confState     raftpb.ConfState
 	snapshotIndex uint64
@@ -64,7 +70,9 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, peers map[uint64]*url
 		nodeID: connectParams.NodeID,
 		node:   node,
 
-		storage: storage,
+		storage:   storage,
+		commitChs: make(map[string]chan EntryCommit),
+		tracker:   newProposalsTracker(),
 	}
 
 	err = consensus.initTransport(ctx, connectParams.NodeID, connectParams.ClusterID, peers)
@@ -82,7 +90,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, peers map[uint64]*url
 	consensus.appliedIndex = snap.Metadata.Index
 
 	go func() {
-		err := consensus.handleReady(connectParams.TickInterval)
+		err := consensus.handleReady(connectParams.TickInterval, connectParams.SnapshotIntervalEntries)
 		if err != nil {
 			consensus.logger.Error("handleReady exited with error, shutting down consensus", zap.Error(err))
 		}
@@ -104,6 +112,31 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, peers map[uint64]*url
 	return consensus, nil
 }
 
+// ProposeValue blocks until the proposal is committed and applied / dropped or until ctx is cancelled.
+func (c *Consensus) ProposeValue(ctx context.Context, proposal Proposal) (any, error) {
+	buf, err := json.Marshal(proposal)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to marshal proposal")
+	}
+
+	applied := c.tracker.track(proposal.ID)
+
+	err = c.node.Propose(ctx, buf)
+	if err != nil {
+		c.tracker.untrack(proposal.ID, ProposalResult{Error: err})
+		return nil, stacktrace.Propagate(err, "failed to propose value to Raft")
+	}
+
+	select {
+	case res := <-applied:
+		return res.Result, res.Error
+
+	case <-ctx.Done():
+		c.tracker.untrack(proposal.ID, ProposalResult{Error: ctx.Err()})
+		return nil, ctx.Err()
+	}
+}
+
 func peersList(peers map[uint64]*url.URL) []raft.Peer {
 	result := make([]raft.Peer, 0, len(peers))
 	for id := range peers {
@@ -162,7 +195,7 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 }
 
 // handleReady processes the Ready channel of the Raft node and applies committed entries to the state machine
-func (c *Consensus) handleReady(tickInterval time.Duration) error {
+func (c *Consensus) handleReady(tickInterval time.Duration, snapshotInterval uint64) error {
 	ticker := time.NewTicker(tickInterval)
 	defer ticker.Stop()
 
@@ -205,7 +238,7 @@ func (c *Consensus) handleReady(tickInterval time.Duration) error {
 				return stacktrace.Propagate(err, "failed to get entries to apply")
 			}
 
-			err = c.publishEntries(entries)
+			err = c.publishEntries(entries, snapshotInterval)
 			if err != nil {
 				return stacktrace.Propagate(err, "failed to publish entries")
 			}
@@ -215,16 +248,135 @@ func (c *Consensus) handleReady(tickInterval time.Duration) error {
 	}
 }
 
-// TODO implement
-func (c *Consensus) publishEntries(_ []raftpb.Entry) error {
+func (c *Consensus) publishEntries(entries []raftpb.Entry, snapshotInterval uint64) error {
+	if len(entries) == 0 {
+		return nil
+	}
+
+	c.logger.Info("publishing entries", zap.Int("numEntries", len(entries)), zap.Uint64("firstIndex", entries[0].Index), zap.Uint64("lastIndex", entries[len(entries)-1].Index))
+
+	var triggerSnapshot bool
+	var err error
+	var wg sync.WaitGroup
+	for _, entry := range entries {
+		switch entry.Type {
+		case raftpb.EntryNormal:
+			err := c.processNormalEntry(entry.Data, &wg)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to process normal entry")
+			}
+		case raftpb.EntryConfChange:
+			err := c.processConfigChangeEntry(entry.Data)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to process config change entry")
+			}
+		case raftpb.EntryConfChangeV2:
+			triggerSnapshot, err = c.processConfigChangeV2Entry(entry.Data)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to process config change v2 entry")
+			}
+		}
+	}
+
+	// wait for all entries to be applied before updating the applied index and potentially triggering a snapshot
+	wg.Wait()
+	c.appliedIndex = entries[len(entries)-1].Index
+
+	if triggerSnapshot || c.appliedIndex-c.snapshotIndex >= snapshotInterval {
+		err := c.storage.triggerSnapshot(c.appliedIndex, &c.confState)
+		if err != nil {
+			return stacktrace.Propagate(err, "failed to trigger snapshot")
+		}
+
+		c.snapshotIndex = c.appliedIndex
+	}
+
 	return nil
 }
 
-// TODO implement
-func (c *Consensus) dispatchSnapshot(_ []byte) error {
+// processNormalEntry passes the proposal to the store and waits for the result to be returned before untracking it.
+func (c *Consensus) processNormalEntry(data []byte, wg *sync.WaitGroup) error {
+	if len(data) <= 0 {
+		return nil
+	}
+
+	prop := Proposal{}
+	err := json.Unmarshal(data, &prop)
+	if err != nil {
+		return stacktrace.Propagate(err, "failed to unmarshal committed proposal")
+	}
+
+	//if readOnly proposal and we did not initiate it, skip it (noop)
+	if prop.ReadOnly && !c.tracker.isPending(prop.ID) {
+		return nil
+	}
+
+	applyDoneC := make(chan ProposalResult, 1)
+	wg.Go(func() {
+		res := <-applyDoneC
+		if c.tracker.isPending(prop.ID) {
+			c.tracker.untrack(prop.ID, res)
+		}
+	})
+
+	ch, ok := c.commitChs[prop.DBName]
+	if !ok {
+		return stacktrace.NewError("no commit channel found for %s", prop.DBName)
+	}
+
+	ch <- EntryCommit{Prop: prop, Done: applyDoneC}
+	return nil
+}
+
+func (c *Consensus) dispatchSnapshot(snapshotData []byte) error {
+	var snapshot map[string][]byte
+	err := json.Unmarshal(snapshotData, &snapshot)
+	if err != nil {
+		return stacktrace.Propagate(err, "failed to unmarshal snapshot data")
+	}
+
+	var eg errgroup.Group
+	for name, data := range snapshot {
+		eg.Go(func() error {
+			ch, ok := c.commitChs[name]
+			if !ok {
+				return stacktrace.NewError("no commit channel found for %s", name)
+			}
+
+			ch <- EntryCommit{SnapshotData: data}
+			return nil
+		})
+	}
+
+	return eg.Wait()
+}
+
+// raftpb.ConfChange is still used internally by Raft, we just need to apply the change to the node.
+// Changes requested by clients are processed by processConfigChangeV2Entry.
+func (c *Consensus) processConfigChangeEntry(data []byte) error {
+	var cc raftpb.ConfChange
+	err := cc.Unmarshal(data)
+	if err != nil {
+		return stacktrace.Propagate(err, "failed to unmarshal config change data")
+	}
+
+	c.confState = *c.node.ApplyConfChange(cc)
 	return nil
 }
 
+func (c *Consensus) processConfigChangeV2Entry(data []byte) (bool, error) {
+	var cc raftpb.ConfChangeV2
+	err := cc.Unmarshal(data)
+	if err != nil {
+		return false, stacktrace.Propagate(err, "failed to unmarshal config change data")
+	}
+
+	c.confState = *c.node.ApplyConfChange(cc)
+
+	// TODO - implement config changes when triggered by a proposal
+	return false, nil
+}
+
 func (c *Consensus) entriesToApply(entries []raftpb.Entry) ([]raftpb.Entry, error) {
 	if len(entries) == 0 {
 		return entries, nil
@@ -276,6 +428,10 @@ func (c *Consensus) ReportSnapshot(id uint64, status raft.SnapshotStatus) {
 }
 
 // RegisterStore allows registering a snapshot provider function for a specific store
-func (c *Consensus) RegisterStore(name string, provider snapshotProvider) {
+// and returns the channel on which committed entries for that store will be sent.
+func (c *Consensus) RegisterStore(name string, provider snapshotProvider) chan EntryCommit {
 	c.storage.registerSnapshotProvider(name, provider)
+	ch := make(chan EntryCommit)
+	c.commitChs[name] = ch
+	return ch
 }
diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go
new file mode 100644
index 000000000..b1243ae30
--- /dev/null
+++ b/pkg/raftstore/consensus/proposal.go
@@ -0,0 +1,65 @@
+package consensus
+
+import (
+	"sync"
+	"time"
+)
+
+type EntryCommit struct {
+	Prop Proposal
+	Done chan ProposalResult
+
+	SnapshotData []byte
+}
+
+type Proposal struct {
+	ID          string            `json:"id"`
+	DBName      string            `json:"dbname"`
+	Timestamp   time.Time         `json:"timestamp"`
+	RequestType string            `json:"request_type"`
+	Value       []byte            `json:"value"`
+	ReadOnly    bool              `json:"read_only"`
+	Parameters  map[string][]byte `json:"parameters,omitempty"`
+}
+
+type ProposalResult struct {
+	Result any
+	Error  error
+}
+
+type proposalsTracker struct {
+	sync.Mutex
+	pending map[string]chan ProposalResult
+}
+
+func newProposalsTracker() *proposalsTracker {
+	return &proposalsTracker{
+		pending: make(map[string]chan ProposalResult),
+	}
+}
+
+func (p *proposalsTracker) isPending(id string) bool {
+	p.Lock()
+	defer p.Unlock()
+
+	_, ok := p.pending[id]
+	return ok
+}
+
+func (p *proposalsTracker) track(id string) chan ProposalResult {
+	p.Lock()
+	defer p.Unlock()
+
+	applied := make(chan ProposalResult, 1)
+	p.pending[id] = applied
+	return applied
+}
+
+func (p *proposalsTracker) untrack(id string, result ProposalResult) {
+	p.Lock()
+	defer p.Unlock()
+
+	applied := p.pending[id]
+	applied <- result
+	delete(p.pending, id)
+}
diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go
index 3ef62c157..f69c181fc 100644
--- a/pkg/raftstore/store.go
+++ b/pkg/raftstore/store.go
@@ -41,7 +41,7 @@ func Init[R any](ctx context.Context, logger *zap.Logger, newRepo func() R) (*St
 	}
 
 	// TODO: implement
-	sharedConsensus.RegisterStore("provider", func() ([]byte, error) {
+	_ = sharedConsensus.RegisterStore("provider", func() ([]byte, error) {
 		return nil, nil
 	})
 

From 5f6437476308b9cb05c1d7b9d241ca3d83fbe8fd Mon Sep 17 00:00:00 2001
From: mahdi-orbitalize <mahdi.atallah@orbitalize.com>
Date: Tue, 16 Jun 2026 11:59:55 +0200
Subject: [PATCH 3/5] add TLS encryption

---
 .gitignore                                   |  5 +-
 build/test-certs/raft-certs/gen-cert.sh      | 60 ++++++++++++++++++++
 build/test-certs/raft-certs/start_cluster.sh | 27 +++++++++
 pkg/raftstore/consensus/consensus.go         | 25 +++++++-
 pkg/raftstore/params/params.go               | 13 +++++
 pkg/raftstore/params/params_test.go          | 31 +++++-----
 6 files changed, 144 insertions(+), 17 deletions(-)
 create mode 100755 build/test-certs/raft-certs/gen-cert.sh
 create mode 100755 build/test-certs/raft-certs/start_cluster.sh

diff --git a/.gitignore b/.gitignore
index b20797585..6e8f25a97 100644
--- a/.gitignore
+++ b/.gitignore
@@ -137,4 +137,7 @@ go
 .terraform*
 
 # raft
-raft_data/
\ No newline at end of file
+raft_data/
+# local certificates
+build/test-certs/raft-certs/*.crt
+build/test-certs/raft-certs/*.key
diff --git a/build/test-certs/raft-certs/gen-cert.sh b/build/test-certs/raft-certs/gen-cert.sh
new file mode 100755
index 000000000..cca5fba49
--- /dev/null
+++ b/build/test-certs/raft-certs/gen-cert.sh
@@ -0,0 +1,60 @@
+#!/bin/bash
+
+# Default values
+NODE_COUNT=3
+
+# Help function
+show_help() {
+  echo "Usage: ./gen-cert.sh [options]"
+  echo ""
+  echo "Options:"
+  echo "  -n, --nodes <num>   Number of nodes to generate (default: 3)"
+  echo "  -c, --clean         Delete all .key, .crt, .csr, and .srl files"
+  echo "  -h, --help          Show this help message"
+  exit 0
+}
+
+# Cleanup function
+clean_certs() {
+    echo "Cleaning up all certificates and keys..."
+    rm -f ./*.key ./*.crt ./*.csr ./*.srl
+    echo "Done."
+    exit 0
+}
+
+# Parse arguments
+while [[ "$#" -gt 0 ]]; do
+  case $1 in
+    -n|--nodes) NODE_COUNT="$2"; shift ;;
+    -c|--clean) clean_certs ;;
+    -h|--help) show_help ;;
+    *) echo "Unknown parameter: $1"; show_help ;;
+  esac
+  shift
+done
+
+echo "Creating CA and certificates for $NODE_COUNT nodes..."
+
+# 1. Generate the Root CA
+openssl req -x509 -newkey rsa:4096 -keyout ca.key -out ca.crt -days 3650 -sha256 -nodes -subj "/CN=root"
+
+# 2. Loop to create node certificates
+for ((i=1; i<=NODE_COUNT; i++))
+do
+    echo "Generating node$i..."
+
+    # Create Request
+    openssl req -out "node$i.csr" -newkey rsa:2048 -keyout "node$i.key" -nodes \
+      -subj "/CN=node$i" \
+      -addext "subjectAltName=DNS:node$i,DNS:localhost,IP:127.0.0.1"
+
+    # Sign Certificate
+    openssl x509 -req -in "node$i.csr" -CA ca.crt -CAkey ca.key \
+      -out "node$i.crt" -days 365 -sha256 \
+      -set_serial "$i" \
+      -extfile <(echo "subjectAltName=DNS:node$i,DNS:localhost,IP:127.0.0.1")
+done
+
+# 3. Final Cleanup of temporary request files
+rm -f ./*.csr
+echo "Success! Generated $NODE_COUNT node certificates."
\ No newline at end of file
diff --git a/build/test-certs/raft-certs/start_cluster.sh b/build/test-certs/raft-certs/start_cluster.sh
new file mode 100755
index 000000000..d49b174ab
--- /dev/null
+++ b/build/test-certs/raft-certs/start_cluster.sh
@@ -0,0 +1,27 @@
+#!/bin/bash
+
+ROOT_DIR="$(git rev-parse --show-toplevel)"
+
+# Ensure the logs directory exists
+mkdir -p logs
+
+PEERS="1=https://127.0.0.1:9021,2=https://127.0.0.1:9022,3=https://127.0.0.1:9023"
+
+echo "=== Starting 3-node raft cluster ==="
+
+for NODE_ID in 1 2 3; do
+    ADDR=":808$NODE_ID"
+    OUT="logs/node$NODE_ID.log"
+
+    echo "Starting node $NODE_ID on port $ADDR..."
+    TLS_ARGS="--raft_tls=ca=$ROOT_DIR/build/test-certs/raft-certs/ca.crt,cert=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.crt,key=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.key"
+
+    go run "$ROOT_DIR"/cmds/core-service \
+        --store_type=raft \
+        --raft_node_id=$NODE_ID \
+        --addr=$ADDR \
+        --raft_peers=$PEERS \
+        "$TLS_ARGS" \
+        --accepted_jwt_audiences=dss \
+        --public_key_files="$ROOT_DIR"/build/test-certs/auth2.pem > "$OUT" 2>&1 &
+done
\ No newline at end of file
diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go
index cca7e53ba..280369a45 100644
--- a/pkg/raftstore/consensus/consensus.go
+++ b/pkg/raftstore/consensus/consensus.go
@@ -2,6 +2,7 @@ package consensus
 
 import (
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -13,9 +14,11 @@ import (
 	"github.com/interuss/dss/pkg/logging"
 	params "github.com/interuss/dss/pkg/raftstore/params"
 	"github.com/interuss/stacktrace"
+	"go.etcd.io/etcd/client/pkg/v3/transport"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
 	v2stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+
 	"go.etcd.io/raft/v3"
 	"go.etcd.io/raft/v3/raftpb"
 	"go.uber.org/zap"
@@ -146,10 +149,19 @@ func peersList(peers map[uint64]*url.URL) []raft.Peer {
 }
 
 func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL) error {
+	connectParams := params.GetConnectParameters()
+
+	tlsInfo := transport.TLSInfo{
+		TrustedCAFile: connectParams.CAFile,
+		CertFile:      connectParams.CertFile,
+		KeyFile:       connectParams.KeyFile,
+	}
+
 	nodeIDStr := fmt.Sprintf("%d", nodeID)
 
 	transport := &rafthttp.Transport{
 		Logger:      logging.WithValuesFromContext(ctx, c.logger.With(zap.String("component", "transport"))),
+		TLSInfo:     tlsInfo,
 		ID:          types.ID(nodeID),
 		ClusterID:   types.ID(clusterID),
 		Raft:        c,
@@ -177,13 +189,20 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		return stacktrace.NewError("node ID %d not found in peers map", nodeID)
 	}
 
+	cfg, err := tlsInfo.ServerConfig()
+	cfg.ClientAuth = tls.RequireAndVerifyClientCert
+	if err != nil {
+		return stacktrace.NewError("failed to create TLS config")
+	}
+
 	c.server = &http.Server{
-		Addr:    listeningAddr,
-		Handler: transport.Handler(),
+		Addr:      listeningAddr,
+		Handler:   transport.Handler(),
+		TLSConfig: cfg,
 	}
 
 	go func() {
-		err := c.server.ListenAndServe()
+		err := c.server.ListenAndServeTLS(tlsInfo.CertFile, tlsInfo.KeyFile)
 		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			c.logger.Error("http server error", zap.Error(err))
 			c.transport.ErrorC <- err
diff --git a/pkg/raftstore/params/params.go b/pkg/raftstore/params/params.go
index 460dcb7b1..205afe28f 100644
--- a/pkg/raftstore/params/params.go
+++ b/pkg/raftstore/params/params.go
@@ -64,6 +64,10 @@ type (
 		MaxSizePerMsg uint64
 		// max number of in-flight messages during optimistic replication phase
 		MaxInflightMsgs int
+
+		CAFile   string
+		CertFile string
+		KeyFile  string
 	}
 )
 
@@ -95,6 +99,10 @@ func (c ConnectParameters) PeerMap() (map[uint64]*url.URL, error) {
 			return nil, stacktrace.Propagate(err, "invalid peer URL %s", parts[1])
 		}
 
+		if peerURL.Scheme != "https" {
+			return nil, stacktrace.NewError("invalid peer URL %s: must use https scheme", parts[1])
+		}
+
 		peers[id] = peerURL
 	}
 
@@ -135,6 +143,11 @@ func init() {
 		"Ticks between leader heartbeats. Effective interval = raft_heartbeat_tick x raft_tick_interval. Must be less than raft_election_tick. Lower values detect follower loss faster but increase network traffic.")
 	flag.Uint64Var(&connectParameters.MaxSizePerMsg, "raft_max_size_per_msg", defaultMaxSizePerMsg, "Maximum bytes in a single Raft message sent to a peer. Smaller values lower the recovery cost but increase the number of messages sent, affecting throughput during replication.")
 	flag.IntVar(&connectParameters.MaxInflightMsgs, "raft_max_inflight_msgs", defaultMaxInflightMsgs, "Maximum number of in-flight Raft messages during optimistic replication phase. This should be set to avoid overflowing the transport layer sending buffer.")
+
+	flag.StringVar(&connectParameters.CAFile, "raft_tls_ca", "", `CA certificate, format: ca=/path/to/ca.crt`)
+	flag.StringVar(&connectParameters.CertFile, "raft_tls_crt", "", `node's certificate, format: ca=/path/to/node.crt`)
+	flag.StringVar(&connectParameters.KeyFile, "raft_tls_key", "", `node's private key, format: key=/path/to/node.key`)
+
 }
 
 // GetConnectParameters returns a ConnectParameters instance that gets populated from well-known CLI flags.
diff --git a/pkg/raftstore/params/params_test.go b/pkg/raftstore/params/params_test.go
index 7410bf64e..f26ea3cff 100644
--- a/pkg/raftstore/params/params_test.go
+++ b/pkg/raftstore/params/params_test.go
@@ -17,22 +17,22 @@ func TestPeerMap(t *testing.T) {
 	}{
 		{
 			name:  "valid single peer",
-			peers: "1=http://node1:9021",
-			want:  map[uint64]*url.URL{1: mustParseURL("http://node1:9021")},
+			peers: "1=https://node1:9021",
+			want:  map[uint64]*url.URL{1: mustParseURL("https://node1:9021")},
 		},
 		{
 			name:  "valid multiple peers",
-			peers: "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021",
+			peers: "1=https://node1:9021,2=https://node2:9021,3=https://node3:9021",
 			want: map[uint64]*url.URL{
-				1: mustParseURL("http://node1:9021"),
-				2: mustParseURL("http://node2:9021"),
-				3: mustParseURL("http://node3:9021"),
+				1: mustParseURL("https://node1:9021"),
+				2: mustParseURL("https://node2:9021"),
+				3: mustParseURL("https://node3:9021"),
 			},
 		},
 		{
 			name:  "valid URL with equals sign in query string",
-			peers: "1=http://node1:9021?token=abc123",
-			want:  map[uint64]*url.URL{1: mustParseURL("http://node1:9021?token=abc123")},
+			peers: "1=https://node1:9021?token=abc123",
+			want:  map[uint64]*url.URL{1: mustParseURL("https://node1:9021?token=abc123")},
 		},
 		{
 			name:      "invalid empty peers string",
@@ -46,27 +46,32 @@ func TestPeerMap(t *testing.T) {
 		},
 		{
 			name:      "invalid non-numeric node ID",
-			peers:     "abc=http://node1:9021",
+			peers:     "abc=https://node1:9021",
 			wantError: true,
 		},
 		{
 			name:      "invalid negative node ID",
-			peers:     "-1=http://node1:9021",
+			peers:     "-1=https://node1:9021",
 			wantError: true,
 		},
 		{
 			name:      "mixed valid and invalid entries",
-			peers:     "1=http://node1:9021,badentry",
+			peers:     "1=https://node1:9021,badentry",
 			wantError: true,
 		},
 		{
 			name:      "invalid zero peer ID",
-			peers:     "0=http://node1:9021",
+			peers:     "0=https://node1:9021",
 			wantError: true,
 		},
 		{
 			name:      "duplicate peer IDs",
-			peers:     "1=http://node1:9021,1=http://node2:9021",
+			peers:     "1=https://node1:9021,1=https://node2:9021",
+			wantError: true,
+		},
+		{
+			name:      "invalid URL scheme",
+			peers:     "1=http://node1:9021",
 			wantError: true,
 		},
 	}

From 9f0ae5532ae13cc33a346ddbcf33bd791c94ec25 Mon Sep 17 00:00:00 2001
From: mahdi-orbitalize <mahdi.atallah@orbitalize.com>
Date: Tue, 16 Jun 2026 12:16:25 +0200
Subject: [PATCH 4/5] fix script to support 3 tls args

---
 build/test-certs/raft-certs/start_cluster.sh | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/build/test-certs/raft-certs/start_cluster.sh b/build/test-certs/raft-certs/start_cluster.sh
index d49b174ab..10ffd813b 100755
--- a/build/test-certs/raft-certs/start_cluster.sh
+++ b/build/test-certs/raft-certs/start_cluster.sh
@@ -14,14 +14,14 @@ for NODE_ID in 1 2 3; do
     OUT="logs/node$NODE_ID.log"
 
     echo "Starting node $NODE_ID on port $ADDR..."
-    TLS_ARGS="--raft_tls=ca=$ROOT_DIR/build/test-certs/raft-certs/ca.crt,cert=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.crt,key=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.key"
-
     go run "$ROOT_DIR"/cmds/core-service \
         --store_type=raft \
         --raft_node_id=$NODE_ID \
         --addr=$ADDR \
         --raft_peers=$PEERS \
-        "$TLS_ARGS" \
+        --raft_tls_ca=$ROOT_DIR/build/test-certs/raft-certs/ca.crt \
+        --raft_tls_crt=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.crt  \
+        --raft_tls_key=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.key \
         --accepted_jwt_audiences=dss \
         --public_key_files="$ROOT_DIR"/build/test-certs/auth2.pem > "$OUT" 2>&1 &
 done
\ No newline at end of file

From 64f3f9a90c1a204174e92dea90e04bf456726a52 Mon Sep 17 00:00:00 2001
From: mahdi-orbitalize <mahdi.atallah@orbitalize.com>
Date: Thu, 18 Jun 2026 16:06:24 +0200
Subject: [PATCH 5/5] add insecure mode for raft transport

---
 pkg/raftstore/consensus/consensus.go | 83 ++++++++++++++++++++--------
 pkg/raftstore/params/params.go       | 15 +++--
 2 files changed, 71 insertions(+), 27 deletions(-)

diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go
index 280369a45..b1f112b75 100644
--- a/pkg/raftstore/consensus/consensus.go
+++ b/pkg/raftstore/consensus/consensus.go
@@ -151,17 +151,21 @@ func peersList(peers map[uint64]*url.URL) []raft.Peer {
 func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL) error {
 	connectParams := params.GetConnectParameters()
 
+	if connectParams.Insecure {
+		c.logger.Warn("Running in insecure mode: HTTPS is disabled or certificates are not verified. Use only for testing or trusted environments.")
+	}
+
 	tlsInfo := transport.TLSInfo{
-		TrustedCAFile: connectParams.CAFile,
-		CertFile:      connectParams.CertFile,
-		KeyFile:       connectParams.KeyFile,
+		TrustedCAFile:      connectParams.CAFile,
+		CertFile:           connectParams.CertFile,
+		KeyFile:            connectParams.KeyFile,
+		InsecureSkipVerify: connectParams.Insecure,
 	}
 
 	nodeIDStr := fmt.Sprintf("%d", nodeID)
 
 	transport := &rafthttp.Transport{
 		Logger:      logging.WithValuesFromContext(ctx, c.logger.With(zap.String("component", "transport"))),
-		TLSInfo:     tlsInfo,
 		ID:          types.ID(nodeID),
 		ClusterID:   types.ID(clusterID),
 		Raft:        c,
@@ -170,8 +174,11 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		ErrorC:      make(chan error),
 	}
 
-	err := transport.Start()
-	if err != nil {
+	if !(connectParams.Insecure && connectParams.CAFile == "" && connectParams.CertFile == "" && connectParams.KeyFile == "") {
+		transport.TLSInfo = tlsInfo
+	}
+
+	if err := transport.Start(); err != nil {
 		return stacktrace.Propagate(err, "failed to start transport")
 	}
 
@@ -181,7 +188,6 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 			listeningAddr = ":" + peerURL.Port()
 			continue
 		}
-
 		transport.AddPeer(types.ID(peerID), []string{peerURL.String()})
 	}
 
@@ -189,28 +195,61 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		return stacktrace.NewError("node ID %d not found in peers map", nodeID)
 	}
 
-	cfg, err := tlsInfo.ServerConfig()
-	cfg.ClientAuth = tls.RequireAndVerifyClientCert
+	var err error
+	c.server, err = c.createServer(listeningAddr, transport.Handler(), tlsInfo, connectParams.Insecure)
 	if err != nil {
-		return stacktrace.NewError("failed to create TLS config")
+		return err
 	}
 
-	c.server = &http.Server{
-		Addr:      listeningAddr,
-		Handler:   transport.Handler(),
-		TLSConfig: cfg,
+	c.transport = transport
+	return nil
+}
+
+func (c *Consensus) createServer(addr string, handler http.Handler, tlsInfo transport.TLSInfo, insecure bool) (*http.Server, error) {
+	server := &http.Server{
+		Addr:    addr,
+		Handler: handler,
 	}
 
-	go func() {
-		err := c.server.ListenAndServeTLS(tlsInfo.CertFile, tlsInfo.KeyFile)
-		if err != nil && !errors.Is(err, http.ErrServerClosed) {
-			c.logger.Error("http server error", zap.Error(err))
-			c.transport.ErrorC <- err
+	if insecure {
+		if tlsInfo.CertFile == "" || tlsInfo.KeyFile == "" {
+			// Certificates provided: no HTTPS
+			go c.startServer(server, false, "", "")
+		} else {
+			// Certificates provided: HTTPS with no verification
+			cfg, err := tlsInfo.ServerConfig()
+			if err != nil {
+				return nil, stacktrace.NewError("failed to create TLS config")
+			}
+			cfg.ClientAuth = tls.NoClientCert
+			server.TLSConfig = cfg
+			go c.startServer(server, true, tlsInfo.CertFile, tlsInfo.KeyFile)
 		}
-	}()
+	} else {
+		// Secure mode: HTTPS with full verification
+		cfg, err := tlsInfo.ServerConfig()
+		if err != nil {
+			return nil, stacktrace.NewError("failed to create TLS config")
+		}
+		cfg.ClientAuth = tls.RequireAndVerifyClientCert
+		server.TLSConfig = cfg
+		go c.startServer(server, true, tlsInfo.CertFile, tlsInfo.KeyFile)
+	}
 
-	c.transport = transport
-	return nil
+	return server, nil
+}
+
+func (c *Consensus) startServer(server *http.Server, useTLS bool, certFile, keyFile string) {
+	var err error
+	if useTLS {
+		err = server.ListenAndServeTLS(certFile, keyFile)
+	} else {
+		err = server.ListenAndServe()
+	}
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+		c.logger.Error("http server error", zap.Error(err))
+		c.transport.ErrorC <- err
+	}
 }
 
 // handleReady processes the Ready channel of the Raft node and applies committed entries to the state machine
diff --git a/pkg/raftstore/params/params.go b/pkg/raftstore/params/params.go
index 205afe28f..3580dad41 100644
--- a/pkg/raftstore/params/params.go
+++ b/pkg/raftstore/params/params.go
@@ -65,6 +65,7 @@ type (
 		// max number of in-flight messages during optimistic replication phase
 		MaxInflightMsgs int
 
+		Insecure bool
 		CAFile   string
 		CertFile string
 		KeyFile  string
@@ -99,10 +100,14 @@ func (c ConnectParameters) PeerMap() (map[uint64]*url.URL, error) {
 			return nil, stacktrace.Propagate(err, "invalid peer URL %s", parts[1])
 		}
 
-		if peerURL.Scheme != "https" {
+		if !c.Insecure && peerURL.Scheme != "https" {
 			return nil, stacktrace.NewError("invalid peer URL %s: must use https scheme", parts[1])
 		}
 
+		if c.Insecure && (peerURL.Scheme != "http" && peerURL.Scheme != "https") {
+			return nil, stacktrace.NewError("invalid peer URL %s: must use http or https scheme", parts[1])
+		}
+
 		peers[id] = peerURL
 	}
 
@@ -144,10 +149,10 @@ func init() {
 	flag.Uint64Var(&connectParameters.MaxSizePerMsg, "raft_max_size_per_msg", defaultMaxSizePerMsg, "Maximum bytes in a single Raft message sent to a peer. Smaller values lower the recovery cost but increase the number of messages sent, affecting throughput during replication.")
 	flag.IntVar(&connectParameters.MaxInflightMsgs, "raft_max_inflight_msgs", defaultMaxInflightMsgs, "Maximum number of in-flight Raft messages during optimistic replication phase. This should be set to avoid overflowing the transport layer sending buffer.")
 
-	flag.StringVar(&connectParameters.CAFile, "raft_tls_ca", "", `CA certificate, format: ca=/path/to/ca.crt`)
-	flag.StringVar(&connectParameters.CertFile, "raft_tls_crt", "", `node's certificate, format: ca=/path/to/node.crt`)
-	flag.StringVar(&connectParameters.KeyFile, "raft_tls_key", "", `node's private key, format: key=/path/to/node.key`)
-
+	flag.BoolVar(&connectParameters.Insecure, "raft_insecure", false, "Enable insecure connection to the Raft cluster. If certificates are provided, they are not verified. If no certificates are provided, HTTPS is disabled entirely.")
+	flag.StringVar(&connectParameters.CAFile, "raft_tls_ca", "", "Path to the CA certificate file (e.g., /path/to/ca.crt). Required if --raft_insecure is not set.")
+	flag.StringVar(&connectParameters.CertFile, "raft_tls_crt", "", "Path to the node's TLS certificate file (e.g., /path/to/node.crt). Required if --raft_insecure is not set.")
+	flag.StringVar(&connectParameters.KeyFile, "raft_tls_key", "", "Path to the node's TLS private key file (e.g., /path/to/node.key). Required if --raft_insecure is not set.")
 }
 
 // GetConnectParameters returns a ConnectParameters instance that gets populated from well-known CLI flags.