func UnsafeClusterDB(c *Cluster) database.DB {
	return c.cluster.DB()
}
UnsafeClusterDB accesses the cluster database, mainly for tests, so that we
don't expose the DB directly on the Cluster.
package db
import (
	"fmt"
	"sync"
	"time"
	"github.com/CanonicalLtd/go-dqlite"
	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/pkg/errors"
	"github.com/spoke-d/thermionic/internal/clock"
	"github.com/spoke-d/thermionic/internal/db/cluster"
	"github.com/spoke-d/thermionic/internal/db/database"
	q "github.com/spoke-d/thermionic/internal/db/query"
)
type ClusterTransactioner interface {
	Transaction(f func(*ClusterTx) error) error
}
type ClusterOpener interface {
	Open(cluster.ServerStore, string, string, time.Duration, ...dqlite.DriverOption) error
}
type ClusterExclusiveLocker interface {
	EnterExclusive() error
	ExitExclusive(func(*ClusterTx) error) error
}
var ErrSomeNodesAreBehind = errors.Errorf("some nodes are behind this node's version")
type QueryCluster interface {
	database.DBAccessor
	Open(cluster.ServerStore, ...dqlite.DriverOption) error
	EnsureSchema(string, string) (bool, error)
	SchemaVersion() int
}
type ClusterTxProvider interface {
	New(database.Tx, int64) *ClusterTx
}
type Cluster struct {
	cluster           QueryCluster  
	clusterTxProvider ClusterTxProvider
	nodeID            int64  
	mu                sync.RWMutex
	transaction       Transaction  
	logger            log.Logger
	clock             clock.Clock
	sleeper           clock.Sleeper
}
func NewCluster(cluster QueryCluster, options ...ClusterOption) *Cluster {
	opts := newClusterOptions()
	for _, option := range options {
		option(opts)
	}
	return &Cluster{
		cluster:           cluster,
		clusterTxProvider: opts.clusterTxProvider,
		transaction:       opts.transaction,
		logger:            opts.logger,
		clock:             opts.clock,
		sleeper:           opts.sleeper,
	}
}
func (c *Cluster) Open(
	store cluster.ServerStore,
	address, dir string,
	timeout time.Duration,
	options ...dqlite.DriverOption,
) error {
	if err := c.cluster.Open(store, options...); err != nil {
		return errors.Wrap(err, "failed to open database")
	}
	after := c.clock.After(timeout)
	for i := 0; ; i++ {
		logPriority := 1  
		if i > 5 {
			logPriority = 2
			if i > 15 && !((i % 5) == 0) {
				logPriority = 0
			}
		}
		err := c.cluster.DB().Ping()
		if err == nil {
			break
		}
		if errors.Cause(err) != dqlite.ErrNoAvailableLeader {
			return errors.WithStack(err)
		}
		msg := fmt.Sprintf("Failed connecting to global database (attempt %d)", i)
		switch logPriority {
		case 1:
			level.Debug(c.logger).Log("err", err, "msg", msg)
		case 2:
			level.Warn(c.logger).Log("err", err, "msg", msg)
		}
		c.sleeper.Sleep(2 * time.Second)
		select {
		case <-after:
			return errors.Errorf("failed to connect to cluster database")
		default:
		}
	}
	nodesVersionsMatch, err := c.cluster.EnsureSchema(address, dir)
	if err != nil {
		return errors.Wrap(err, "failed to ensure schema")
	}
	if err := c.Transaction(func(tx *ClusterTx) error {
		nodes, err := tx.Nodes()
		if err != nil {
			return errors.Wrap(err, "failed to fetch nodes")
		}
		if len(nodes) == 1 && nodes[0].Address == "0.0.0.0" {
			c.NodeID(1)
			return nil
		}
		for _, node := range nodes {
			if node.Address == address {
				c.nodeID = node.ID
				return nil
			}
		}
		return errors.Errorf("no node registered with address %q", address)
	}); err != nil {
		return errors.WithStack(err)
	}
	if !nodesVersionsMatch {
		err = ErrSomeNodesAreBehind
	}
	return err
}
func (c *Cluster) Transaction(f func(*ClusterTx) error) error {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.tx(f)
}
func (c *Cluster) NodeID(id int64) {
	c.nodeID = id
}
func (c *Cluster) DB() database.DB {
	return c.cluster.DB()
}
func (c *Cluster) SchemaVersion() int {
	return c.cluster.SchemaVersion()
}
func (c *Cluster) Close() error {
	return c.cluster.DB().Close()
}
func (c *Cluster) EnterExclusive() error {
	level.Debug(c.logger).Log("msg", "Acquiring exclusive lock on cluster db")
	ch := make(chan struct{})
	go func() {
		c.mu.Lock()
		ch <- struct{}{}
	}()
	timeout := 20 * time.Second
	select {
	case <-ch:
		return nil
	case <-time.After(timeout):
		return errors.Errorf("timeout (%s)", timeout)
	}
}
func (c *Cluster) ExitExclusive(f func(*ClusterTx) error) error {
	level.Debug(c.logger).Log("msg", "Releasing exclusive lock on cluster db")
	defer c.mu.Unlock()
	return c.tx(f)
}
func (c *Cluster) tx(f func(*ClusterTx) error) error {
	nodeID := c.nodeID
	return q.Retry(c.sleeper, func() error {
		return c.transaction.Transaction(c.cluster.DB(), func(tx database.Tx) error {
			return f(c.clusterTxProvider.New(tx, nodeID))
		})
	})
}