func (s *Store) createOnDisk(b []byte) (*sql.DB, error) {
	if err := os.Remove(s.dbPath); err != nil && !os.IsNotExist(err) {
		return nil, err
	}
	if b != nil {
		if err := ioutil.WriteFile(s.dbPath, b, 0660); err != nil {
			return nil, err
		}
	}
	return sql.Open(s.dbPath, s.dbConf.FKConstraints)
}
createOnDisk opens an on-disk database file at the Store's configured path. If
b is non-nil, any preexisting file will first be overwritten with those contents.
Otherwise any pre-existing file will be removed before the database is opened.
package store
import (
	"bytes"
	"compress/gzip"
	"encoding/binary"
	"errors"
	"expvar"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"sync"
	"time"
	"unsafe"
	"github.com/hashicorp/raft"
	"github.com/rqlite/rqlite/command"
	sql "github.com/rqlite/rqlite/db"
	rlog "github.com/rqlite/rqlite/log"
)
var (
	ErrNotLeader = errors.New("not leader")
	ErrStaleRead = errors.New("stale read")
	ErrOpenTimeout = errors.New("timeout waiting for initial logs application")
	ErrInvalidBackupFormat = errors.New("invalid backup format")
)
const (
	raftDBPath          = "raft.db"  
	peersPath           = "raft/peers.json"
	peersInfoPath       = "raft/peers.info"
	retainSnapshotCount = 2
	applyTimeout        = 10 * time.Second
	openTimeout         = 120 * time.Second
	sqliteFile          = "db.sqlite"
	leaderWaitDelay     = 100 * time.Millisecond
	appliedWaitDelay    = 100 * time.Millisecond
	connectionPoolCount = 5
	connectionTimeout   = 10 * time.Second
	raftLogCacheSize    = 512
	trailingScale       = 1.25
)
const (
	numSnaphots              = "num_snapshots"
	numBackups               = "num_backups"
	numRestores              = "num_restores"
	numRecoveries            = "num_recoveries"
	numUncompressedCommands  = "num_uncompressed_commands"
	numCompressedCommands    = "num_compressed_commands"
	numJoins                 = "num_joins"
	numIgnoredJoins          = "num_ignored_joins"
	numRemovedBeforeJoins    = "num_removed_before_joins"
	snapshotCreateDuration   = "snapshot_create_duration"
	snapshotPersistDuration  = "snapshot_persist_duration"
	snapshotDBSerializedSize = "snapshot_db_serialized_size"
	snapshotDBOnDiskSize     = "snapshot_db_ondisk_size"
)
type BackupFormat int
const (
	BackupSQL BackupFormat = iota
	BackupBinary
)
var stats *expvar.Map
func init() {
	stats = expvar.NewMap("store")
	stats.Add(numSnaphots, 0)
	stats.Add(numBackups, 0)
	stats.Add(numRestores, 0)
	stats.Add(numRecoveries, 0)
	stats.Add(numUncompressedCommands, 0)
	stats.Add(numCompressedCommands, 0)
	stats.Add(numJoins, 0)
	stats.Add(numIgnoredJoins, 0)
	stats.Add(numRemovedBeforeJoins, 0)
	stats.Add(snapshotCreateDuration, 0)
	stats.Add(snapshotPersistDuration, 0)
	stats.Add(snapshotDBSerializedSize, 0)
	stats.Add(snapshotDBOnDiskSize, 0)
}
type ClusterState int
const (
	Leader ClusterState = iota
	Follower
	Candidate
	Shutdown
	Unknown
)
type Store struct {
	raftDir       string
	peersPath     string
	peersInfoPath string
	raft   *raft.Raft  
	ln     Listener
	raftTn *raft.NetworkTransport
	raftID string     
	dbConf *DBConfig  
	dbPath string     
	db     *sql.DB    
	queryTxMu sync.RWMutex
	dbAppliedIndexMu sync.RWMutex
	dbAppliedIndex   uint64
	fsmIndex   uint64
	fsmIndexMu sync.RWMutex
	reqMarshaller *command.RequestMarshaler  
	raftLog       raft.LogStore              
	raftStable    raft.StableStore           
	boltStore     *rlog.Log                  
	onDiskCreated        bool       
	snapsExistOnOpen     bool       
	firstIdxOnOpen       uint64     
	lastIdxOnOpen        uint64     
	lastCommandIdxOnOpen uint64     
	firstLogAppliedT     time.Time  
	appliedOnOpen        uint64     
	openT                time.Time  
	logger *log.Logger
	ShutdownOnRemove   bool
	SnapshotThreshold  uint64
	SnapshotInterval   time.Duration
	LeaderLeaseTimeout time.Duration
	HeartbeatTimeout   time.Duration
	ElectionTimeout    time.Duration
	ApplyTimeout       time.Duration
	RaftLogLevel       string
	numTrailingLogs uint64
	numNoops       int
	numSnapshotsMu sync.Mutex
	numSnapshots   int
}
func IsNewNode(raftDir string) bool {
	return !pathExists(filepath.Join(raftDir, raftDBPath))
}
type Config struct {
	DBConf *DBConfig    
	Dir    string       
	Tn     Transport    
	ID     string       
	Logger *log.Logger  
}
func New(ln Listener, c *Config) *Store {
	logger := c.Logger
	if logger == nil {
		logger = log.New(os.Stderr, "[store] ", log.LstdFlags)
	}
	dbPath := filepath.Join(c.Dir, sqliteFile)
	if c.DBConf.OnDiskPath != "" {
		dbPath = c.DBConf.OnDiskPath
	}
	return &Store{
		ln:            ln,
		raftDir:       c.Dir,
		peersPath:     filepath.Join(c.Dir, peersPath),
		peersInfoPath: filepath.Join(c.Dir, peersInfoPath),
		raftID:        c.ID,
		dbConf:        c.DBConf,
		dbPath:        dbPath,
		reqMarshaller: command.NewRequestMarshaler(),
		logger:        logger,
		ApplyTimeout:  applyTimeout,
	}
}
func (s *Store) Open(enableBootstrap bool) error {
	s.openT = time.Now()
	s.logger.Printf("opening store with node ID %s", s.raftID)
	if !s.dbConf.Memory {
		s.logger.Printf("configured for an on-disk database at %s", s.dbPath)
		parentDir := filepath.Dir(s.dbPath)
		s.logger.Printf("ensuring directory at %s exists", parentDir)
		err := os.MkdirAll(parentDir, 0755)
		if err != nil {
			return err
		}
	} else {
		s.logger.Printf("configured for an in-memory database")
	}
	s.logger.Printf("ensuring directory at %s exists", s.raftDir)
	err := os.MkdirAll(filepath.Dir(s.peersPath), 0755)
	if err != nil {
		return err
	}
	s.raftTn = raft.NewNetworkTransport(NewTransport(s.ln), connectionPoolCount, connectionTimeout, nil)
	s.numTrailingLogs = uint64(float64(s.SnapshotThreshold) * trailingScale)
	config := s.raftConfig()
	config.LocalID = raft.ServerID(s.raftID)
	snapshots, err := raft.NewFileSnapshotStore(s.raftDir, retainSnapshotCount, os.Stderr)
	if err != nil {
		return fmt.Errorf("file snapshot store: %s", err)
	}
	snaps, err := snapshots.List()
	if err != nil {
		return fmt.Errorf("list snapshots: %s", err)
	}
	s.logger.Printf("%d preexisting snapshots present", len(snaps))
	s.snapsExistOnOpen = len(snaps) > 0
	s.boltStore, err = rlog.NewLog(filepath.Join(s.raftDir, raftDBPath))
	if err != nil {
		return fmt.Errorf("new log store: %s", err)
	}
	s.raftStable = s.boltStore
	s.raftLog, err = raft.NewLogCache(raftLogCacheSize, s.boltStore)
	if err != nil {
		return fmt.Errorf("new cached store: %s", err)
	}
	if pathExists(s.peersPath) {
		s.logger.Printf("attempting node recovery using %s", s.peersPath)
		config, err := raft.ReadConfigJSON(s.peersPath)
		if err != nil {
			return fmt.Errorf("failed to read peers file: %s", err.Error())
		}
		if err = RecoverNode(s.raftDir, s.logger, s.raftLog, s.raftStable, snapshots, s.raftTn, config); err != nil {
			return fmt.Errorf("failed to recover node: %s", err.Error())
		}
		if err := os.Rename(s.peersPath, s.peersInfoPath); err != nil {
			return fmt.Errorf("failed to move %s after recovery: %s", s.peersPath, err.Error())
		}
		s.logger.Printf("node recovered successfully using %s", s.peersPath)
		stats.Add(numRecoveries, 1)
	}
	if err := s.setLogInfo(); err != nil {
		return fmt.Errorf("set log info: %s", err)
	}
	s.logger.Printf("first log index: %d, last log index: %d, last command log index: %d:",
		s.firstIdxOnOpen, s.lastIdxOnOpen, s.lastCommandIdxOnOpen)
	if !s.dbConf.Memory && !s.snapsExistOnOpen && s.lastCommandIdxOnOpen == 0 {
		s.db, err = s.createOnDisk(nil)
		if err != nil {
			return fmt.Errorf("failed to create on-disk database")
		}
		s.onDiskCreated = true
		s.logger.Printf("created on-disk database at open")
	} else {
		s.db, err = s.createInMemory(nil)
		if err != nil {
			return fmt.Errorf("failed to create in-memory database")
		}
		s.logger.Printf("created in-memory database at open")
	}
	ra, err := raft.NewRaft(config, s, s.raftLog, s.raftStable, snapshots, s.raftTn)
	if err != nil {
		return fmt.Errorf("new raft: %s", err)
	}
	if enableBootstrap {
		s.logger.Printf("executing new cluster bootstrap")
		configuration := raft.Configuration{
			Servers: []raft.Server{
				{
					ID:      config.LocalID,
					Address: s.raftTn.LocalAddr(),
				},
			},
		}
		ra.BootstrapCluster(configuration)
	} else {
		s.logger.Printf("no cluster bootstrap requested")
	}
	s.raft = ra
	return nil
}
func (s *Store) Close(wait bool) error {
	f := s.raft.Shutdown()
	if wait {
		if e := f.(raft.Future); e.Error() != nil {
			return e.Error()
		}
	}
	if err := s.db.Close(); err != nil {
		return err
	}
	if err := s.boltStore.Close(); err != nil {
		return err
	}
	return nil
}
func (s *Store) WaitForAppliedFSM(timeout time.Duration) (uint64, error) {
	if timeout == 0 {
		return 0, nil
	}
	return s.WaitForFSMIndex(s.raft.AppliedIndex(), timeout)
}
func (s *Store) WaitForInitialLogs(timeout time.Duration) error {
	if timeout == 0 {
		return nil
	}
	s.logger.Printf("waiting for up to %s for application of initial logs (lcIdx=%d)",
		timeout, s.lastCommandIdxOnOpen)
	return s.WaitForApplied(timeout)
}
func (s *Store) WaitForApplied(timeout time.Duration) error {
	if timeout == 0 {
		return nil
	}
	return s.WaitForAppliedIndex(s.raft.LastIndex(), timeout)
}
func (s *Store) WaitForAppliedIndex(idx uint64, timeout time.Duration) error {
	tck := time.NewTicker(appliedWaitDelay)
	defer tck.Stop()
	tmr := time.NewTimer(timeout)
	defer tmr.Stop()
	for {
		select {
		case <-tck.C:
			if s.raft.AppliedIndex() >= idx {
				return nil
			}
		case <-tmr.C:
			return fmt.Errorf("timeout expired")
		}
	}
}
func (s *Store) IsLeader() bool {
	return s.raft.State() == raft.Leader
}
func (s *Store) State() ClusterState {
	state := s.raft.State()
	switch state {
	case raft.Leader:
		return Leader
	case raft.Candidate:
		return Candidate
	case raft.Follower:
		return Follower
	case raft.Shutdown:
		return Shutdown
	default:
		return Unknown
	}
}
func (s *Store) Path() string {
	return s.raftDir
}
func (s *Store) Addr() string {
	return string(s.raftTn.LocalAddr())
}
func (s *Store) ID() string {
	return s.raftID
}
func (s *Store) LeaderAddr() (string, error) {
	return string(s.raft.Leader()), nil
}
func (s *Store) LeaderID() (string, error) {
	addr, err := s.LeaderAddr()
	if err != nil {
		return "", nil
	}
	configFuture := s.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		s.logger.Printf("failed to get raft configuration: %v", err)
		return "", err
	}
	for _, srv := range configFuture.Configuration().Servers {
		if srv.Address == raft.ServerAddress(addr) {
			return string(srv.ID), nil
		}
	}
	return "", nil
}
func (s *Store) Nodes() ([]*Server, error) {
	f := s.raft.GetConfiguration()
	if f.Error() != nil {
		return nil, f.Error()
	}
	rs := f.Configuration().Servers
	servers := make([]*Server, len(rs))
	for i := range rs {
		servers[i] = &Server{
			ID:       string(rs[i].ID),
			Addr:     string(rs[i].Address),
			Suffrage: rs[i].Suffrage.String(),
		}
	}
	sort.Sort(Servers(servers))
	return servers, nil
}
func (s *Store) WaitForLeader(timeout time.Duration) (string, error) {
	tck := time.NewTicker(leaderWaitDelay)
	defer tck.Stop()
	tmr := time.NewTimer(timeout)
	defer tmr.Stop()
	for {
		select {
		case <-tck.C:
			l, err := s.LeaderAddr()
			if err != nil {
				return "", nil
			}
			if l != "" {
				return l, nil
			}
		case <-tmr.C:
			return "", fmt.Errorf("timeout expired")
		}
	}
}
func (s *Store) SetRequestCompression(batch, size int) {
	s.reqMarshaller.BatchThreshold = batch
	s.reqMarshaller.SizeThreshold = size
}
func (s *Store) WaitForFSMIndex(idx uint64, timeout time.Duration) (uint64, error) {
	tck := time.NewTicker(appliedWaitDelay)
	defer tck.Stop()
	tmr := time.NewTimer(timeout)
	defer tmr.Stop()
	var fsmIdx uint64
	for {
		select {
		case <-tck.C:
			s.fsmIndexMu.RLock()
			fsmIdx = s.fsmIndex
			s.fsmIndexMu.RUnlock()
			if fsmIdx >= idx {
				return fsmIdx, nil
			}
		case <-tmr.C:
			return 0, fmt.Errorf("timeout expired")
		}
	}
}
func (s *Store) Stats() (map[string]interface{}, error) {
	fsmIdx := func() uint64 {
		s.fsmIndexMu.RLock()
		defer s.fsmIndexMu.RUnlock()
		return s.fsmIndex
	}()
	dbAppliedIdx := func() uint64 {
		s.dbAppliedIndexMu.Lock()
		defer s.dbAppliedIndexMu.Unlock()
		return s.dbAppliedIndex
	}()
	dbStatus, err := s.db.Stats()
	if err != nil {
		return nil, err
	}
	nodes, err := s.Nodes()
	if err != nil {
		return nil, err
	}
	leaderID, err := s.LeaderID()
	if err != nil {
		return nil, err
	}
	raftStats := make(map[string]interface{})
	for k, v := range s.raft.Stats() {
		if s, err := strconv.ParseInt(v, 10, 64); err != nil {
			raftStats[k] = v
		} else {
			raftStats[k] = s
		}
	}
	raftStats["log_size"], err = s.logSize()
	if err != nil {
		return nil, err
	}
	raftStats["bolt"] = s.boltStore.Stats()
	dirSz, err := dirSize(s.raftDir)
	if err != nil {
		return nil, err
	}
	leaderAddr, err := s.LeaderAddr()
	if err != nil {
		return nil, err
	}
	status := map[string]interface{}{
		"node_id":          s.raftID,
		"raft":             raftStats,
		"fsm_index":        fsmIdx,
		"db_applied_index": dbAppliedIdx,
		"addr":             s.Addr(),
		"leader": map[string]string{
			"node_id": leaderID,
			"addr":    leaderAddr,
		},
		"apply_timeout":      s.ApplyTimeout.String(),
		"heartbeat_timeout":  s.HeartbeatTimeout.String(),
		"election_timeout":   s.ElectionTimeout.String(),
		"snapshot_threshold": s.SnapshotThreshold,
		"snapshot_interval":  s.SnapshotInterval,
		"trailing_logs":      s.numTrailingLogs,
		"request_marshaler":  s.reqMarshaller.Stats(),
		"nodes":              nodes,
		"dir":                s.raftDir,
		"dir_size":           dirSz,
		"sqlite3":            dbStatus,
		"db_conf":            s.dbConf,
	}
	return status, nil
}
func (s *Store) Execute(ex *command.ExecuteRequest) ([]*command.ExecuteResult, error) {
	if s.raft.State() != raft.Leader {
		return nil, ErrNotLeader
	}
	return s.execute(ex)
}
func (s *Store) execute(ex *command.ExecuteRequest) ([]*command.ExecuteResult, error) {
	b, compressed, err := s.reqMarshaller.Marshal(ex)
	if err != nil {
		return nil, err
	}
	if compressed {
		stats.Add(numCompressedCommands, 1)
	} else {
		stats.Add(numUncompressedCommands, 1)
	}
	c := &command.Command{
		Type:       command.Command_COMMAND_TYPE_EXECUTE,
		SubCommand: b,
		Compressed: compressed,
	}
	b, err = command.Marshal(c)
	if err != nil {
		return nil, err
	}
	af := s.raft.Apply(b, s.ApplyTimeout).(raft.ApplyFuture)
	if af.Error() != nil {
		if af.Error() == raft.ErrNotLeader {
			return nil, ErrNotLeader
		}
		return nil, af.Error()
	}
	s.dbAppliedIndexMu.Lock()
	s.dbAppliedIndex = af.Index()
	s.dbAppliedIndexMu.Unlock()
	r := af.Response().(*fsmExecuteResponse)
	return r.results, r.error
}
func (s *Store) Query(qr *command.QueryRequest) ([]*command.QueryRows, error) {
	if qr.Level == command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG {
		if s.raft.State() != raft.Leader {
			return nil, ErrNotLeader
		}
		b, compressed, err := s.reqMarshaller.Marshal(qr)
		if err != nil {
			return nil, err
		}
		if compressed {
			stats.Add(numCompressedCommands, 1)
		} else {
			stats.Add(numUncompressedCommands, 1)
		}
		c := &command.Command{
			Type:       command.Command_COMMAND_TYPE_QUERY,
			SubCommand: b,
			Compressed: compressed,
		}
		b, err = command.Marshal(c)
		if err != nil {
			return nil, err
		}
		af := s.raft.Apply(b, s.ApplyTimeout).(raft.ApplyFuture)
		if af.Error() != nil {
			if af.Error() == raft.ErrNotLeader {
				return nil, ErrNotLeader
			}
			return nil, af.Error()
		}
		s.dbAppliedIndexMu.Lock()
		s.dbAppliedIndex = af.Index()
		s.dbAppliedIndexMu.Unlock()
		r := af.Response().(*fsmQueryResponse)
		return r.rows, r.error
	}
	if qr.Level == command.QueryRequest_QUERY_REQUEST_LEVEL_WEAK && s.raft.State() != raft.Leader {
		return nil, ErrNotLeader
	}
	if qr.Level == command.QueryRequest_QUERY_REQUEST_LEVEL_NONE && qr.Freshness > 0 &&
		time.Since(s.raft.LastContact()).Nanoseconds() > qr.Freshness {
		return nil, ErrStaleRead
	}
	if qr.Request.Transaction {
		s.queryTxMu.RLock()
		defer s.queryTxMu.RUnlock()
	}
	return s.db.Query(qr.Request, qr.Timings)
}
func (s *Store) Backup(leader bool, fmt BackupFormat, dst io.Writer) error {
	if leader && s.raft.State() != raft.Leader {
		return ErrNotLeader
	}
	if fmt == BackupBinary {
		if err := s.database(leader, dst); err != nil {
			return err
		}
	} else if fmt == BackupSQL {
		if err := s.db.Dump(dst); err != nil {
			return err
		}
	} else {
		return ErrInvalidBackupFormat
	}
	stats.Add(numBackups, 1)
	return nil
}
func (s *Store) Join(id, addr string, voter bool) error {
	s.logger.Printf("received request from node with ID %s, at %s, to join this node", id, addr)
	if s.raft.State() != raft.Leader {
		return ErrNotLeader
	}
	configFuture := s.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		s.logger.Printf("failed to get raft configuration: %v", err)
		return err
	}
	for _, srv := range configFuture.Configuration().Servers {
		if srv.ID == raft.ServerID(id) || srv.Address == raft.ServerAddress(addr) {
			if srv.Address == raft.ServerAddress(addr) && srv.ID == raft.ServerID(id) {
				stats.Add(numIgnoredJoins, 1)
				s.logger.Printf("node %s at %s already member of cluster, ignoring join request", id, addr)
				return nil
			}
			if err := s.remove(id); err != nil {
				s.logger.Printf("failed to remove node %s: %v", id, err)
				return err
			}
			stats.Add(numRemovedBeforeJoins, 1)
			s.logger.Printf("removed node %s prior to rejoin with changed ID or address", id)
		}
	}
	var f raft.IndexFuture
	if voter {
		f = s.raft.AddVoter(raft.ServerID(id), raft.ServerAddress(addr), 0, 0)
	} else {
		f = s.raft.AddNonvoter(raft.ServerID(id), raft.ServerAddress(addr), 0, 0)
	}
	if e := f.(raft.Future); e.Error() != nil {
		if e.Error() == raft.ErrNotLeader {
			return ErrNotLeader
		}
		return e.Error()
	}
	stats.Add(numJoins, 1)
	s.logger.Printf("node with ID %s, at %s, joined successfully as %s", id, addr, prettyVoter(voter))
	return nil
}
func (s *Store) Remove(id string) error {
	s.logger.Printf("received request to remove node %s", id)
	if err := s.remove(id); err != nil {
		s.logger.Printf("failed to remove node %s: %s", id, err.Error())
		return err
	}
	s.logger.Printf("node %s removed successfully", id)
	return nil
}
func (s *Store) Noop(id string) error {
	n := &command.Noop{
		Id: id,
	}
	b, err := command.MarshalNoop(n)
	if err != nil {
		return err
	}
	c := &command.Command{
		Type:       command.Command_COMMAND_TYPE_NOOP,
		SubCommand: b,
	}
	bc, err := command.Marshal(c)
	if err != nil {
		return err
	}
	af := s.raft.Apply(bc, s.ApplyTimeout).(raft.ApplyFuture)
	if af.Error() != nil {
		if af.Error() == raft.ErrNotLeader {
			return ErrNotLeader
		}
		return af.Error()
	}
	return nil
}
func (s *Store) createInMemory(b []byte) (db *sql.DB, err error) {
	if b == nil || len(b) == 0 {
		db, err = sql.OpenInMemory(s.dbConf.FKConstraints)
	} else {
		db, err = sql.DeserializeIntoMemory(b, s.dbConf.FKConstraints)
	}
	return
}