func raftInstanceInit(
	db *db.Node, node *db.RaftNode, cert *shared.CertInfo, latency float64) (*raftInstance, error) {
	// FIXME: should be a parameter
	timeout := 5 * time.Second

	raftLogger := raftLogger()

	// Raft config.
	config := raftConfig(latency)
	config.Logger = raftLogger
	config.LocalID = raft.ServerID(strconv.Itoa(int(node.ID)))

	// Raft transport
	var handler *rafthttp.Handler
	var membershipChanger func(*raft.Raft)
	var layer *rafthttp.Layer
	var transport raft.Transport
	addr := node.Address
	if addr == "" {
		// This should normally be used only for testing as it can
		// cause split-brian, but since we are not exposing raft to the
		// network at all it's safe to do so. When this node gets
		// exposed to the network and assigned an address, we need to
		// restart raft anyways.
		config.StartAsLeader = true
		transport = raftMemoryTransport()
	} else {
		dial, err := raftDial(cert)
		if err != nil {
			return nil, err
		}

		transport, handler, layer, err = raftNetworkTransport(db, addr, log.New(&raftLogWriter{}, "", 0), timeout, dial)
		if err != nil {
			return nil, err
		}
		membershipChanger = func(raft *raft.Raft) {
			raftmembership.HandleChangeRequests(raft, handler.Requests())
		}
	}

	err := raft.ValidateConfig(config)
	if err != nil {
		return nil, errors.Wrap(err, "invalid raft configuration")
	}

	// Rename legacy data directory if needed.
	dir := filepath.Join(db.Dir(), "global")
	legacyDir := filepath.Join(db.Dir(), "..", "raft")
	if shared.PathExists(legacyDir) {
		if shared.PathExists(dir) {
			return nil, fmt.Errorf("both legacy and new global database directories exist")
		}
		logger.Info("Renaming global database directory from raft/ to database/global/")
		err := os.Rename(legacyDir, dir)
		if err != nil {
			return nil, errors.Wrap(err, "failed to rename legacy global database directory")
		}
	}

	// Data directory
	if !shared.PathExists(dir) {
		err := os.Mkdir(dir, 0750)
		if err != nil {
			return nil, err
		}
	}

	// Raft logs store
	logs, err := raftboltdb.New(raftboltdb.Options{
		Path:        filepath.Join(dir, "logs.db"),
		BoltOptions: &bolt.Options{Timeout: timeout},
	})
	if err != nil {
		return nil, errors.Wrap(err, "failed to create bolt store for raft logs")
	}

	// Raft snapshot store (don't log snapshots since we take them frequently)
	snaps, err := raft.NewFileSnapshotStoreWithLogger(dir, 2, log.New(ioutil.Discard, "", 0))
	if err != nil {
		return nil, errors.Wrap(err, "failed to create file snapshot store")
	}

	// If we are the initial node, we use the last index persisted in the
	// logs store and other checks to determine if we have ever
	// bootstrapped the cluster, and if not we do so (see raft.HasExistingState).
	if node.ID == 1 {
		err := raftMaybeBootstrap(config, logs, snaps, transport)
		if err != nil {
			return nil, errors.Wrap(err, "failed to boostrap cluster")
		}
	}

	// The dqlite registry and FSM.
	registry := dqlite.NewRegistry(strconv.Itoa(serial))
	serial++
	fsm := dqlite.NewFSM(registry)

	// The actual raft instance.
	raft, err := raft.NewRaft(config, fsm, logs, logs, snaps, transport)
	if err != nil {
		logs.Close()
		return nil, errors.Wrap(err, "failed to start raft")
	}

	if membershipChanger != nil {
		// Process Raft connections over HTTP. This goroutine will
		// terminate when instance.handler.Close() is called, which
		// happens indirectly when the raft instance is shutdown in
		// instance.Shutdown(), and the associated transport is closed.
		go membershipChanger(raft)
	}

	instance := &raftInstance{
		layer:             layer,
		handler:           raftHandler(cert, handler),
		membershipChanger: membershipChanger,
		logs:              logs,
		registry:          registry,
		fsm:               fsm,
		raft:              raft,
	}

	return instance, nil
}
func raftInstanceInit(
	db *db.Node, node *db.RaftNode, cert *shared.CertInfo, latency float64) (*raftInstance, error) {
	// FIXME: should be a parameter
	timeout := 5 * time.Second

	raftLogger := raftLogger()

	// Raft config.
	config := raftConfig(latency)
	config.Logger = raftLogger
	config.LocalID = raft.ServerID(strconv.Itoa(int(node.ID)))

	// Raft transport
	var handler *rafthttp.Handler
	var membershipChanger func(*raft.Raft)
	var layer *rafthttp.Layer
	var transport raft.Transport
	addr := node.Address
	if addr == "" {
		// This should normally be used only for testing as it can
		// cause split-brian, but since we are not exposing raft to the
		// network at all it's safe to do so. When this node gets
		// exposed to the network and assigned an address, we need to
		// restart raft anyways.
		config.StartAsLeader = true
		transport = raftMemoryTransport()
	} else {
		dial, err := raftDial(cert)
		if err != nil {
			return nil, err
		}

		transport, handler, layer, err = raftNetworkTransport(db, addr, log.New(&raftLogWriter{}, "", 0), timeout, dial)
		if err != nil {
			return nil, err
		}
		membershipChanger = func(raft *raft.Raft) {
			raftmembership.HandleChangeRequests(raft, handler.Requests())
		}
	}

	err := raft.ValidateConfig(config)
	if err != nil {
		return nil, errors.Wrap(err, "invalid raft configuration")
	}

	// Rename legacy data directory if needed.
	dir := filepath.Join(db.Dir(), "global")
	legacyDir := filepath.Join(db.Dir(), "..", "raft")
	if shared.PathExists(legacyDir) {
		if shared.PathExists(dir) {
			return nil, fmt.Errorf("both legacy and new global database directories exist")
		}
		logger.Info("Renaming global database directory from raft/ to database/global/")
		err := os.Rename(legacyDir, dir)
		if err != nil {
			return nil, errors.Wrap(err, "failed to rename legacy global database directory")
		}
	}

	// Data directory
	if !shared.PathExists(dir) {
		err := os.Mkdir(dir, 0750)
		if err != nil {
			return nil, err
		}
	}

	// Raft logs store
	logs, err := raftboltdb.New(raftboltdb.Options{
		Path:        filepath.Join(dir, "logs.db"),
		BoltOptions: &bolt.Options{Timeout: timeout},
	})
	if err != nil {
		return nil, errors.Wrap(err, "failed to create bolt store for raft logs")
	}

	// Raft snapshot store (don't log snapshots since we take them frequently)
	snaps, err := raft.NewFileSnapshotStoreWithLogger(dir, 2, log.New(ioutil.Discard, "", 0))
	if err != nil {
		return nil, errors.Wrap(err, "failed to create file snapshot store")
	}

	// If we are the initial node, we use the last index persisted in the
	// logs store and other checks to determine if we have ever
	// bootstrapped the cluster, and if not we do so (see raft.HasExistingState).
	if node.ID == 1 {
		err := raftMaybeBootstrap(config, logs, snaps, transport)
		if err != nil {
			return nil, errors.Wrap(err, "failed to boostrap cluster")
		}
	}

	// The dqlite registry and FSM.
	registry := dqlite.NewRegistry(strconv.Itoa(serial))
	serial++
	fsm := dqlite.NewFSM(registry)

	// The actual raft instance.
	raft, err := raft.NewRaft(config, fsm, logs, logs, snaps, transport)
	if err != nil {
		logs.Close()
		return nil, errors.Wrap(err, "failed to start raft")
	}

	if membershipChanger != nil {
		// Process Raft connections over HTTP. This goroutine will
		// terminate when instance.handler.Close() is called, which
		// happens indirectly when the raft instance is shutdown in
		// instance.Shutdown(), and the associated transport is closed.
		go membershipChanger(raft)
	}

	instance := &raftInstance{
		layer:             layer,
		handler:           raftHandler(cert, handler),
		membershipChanger: membershipChanger,
		logs:              logs,
		registry:          registry,
		fsm:               fsm,
		raft:              raft,
	}

	return instance, nil
}
// Create a new raftFactory, instantiating all needed raft dependencies.