implement transaction API layer and intra-cluster messaging

also adds a "noSleep" option to the server command to avoid the 5
second sleep we introduced on startup for non-coordinator cluster
nodes. The sleep doesn't seem to be needed in the tests and makes them
much slower.

api.go

@@ -1585,6 +1585,22 @@ func (api *API) PrimaryReplicaNodeURL() url.URL {
 	return node.URI.URL()
 }
 
+func (api *API) StartTransaction(id string, timeout time.Duration, exclusive bool, remote bool) (Transaction, error) {
+	return api.server.StartTransaction(id, timeout, exclusive, remote)
+}
+
+func (api *API) FinishTransaction(id string, remote bool) (Transaction, error) {
+	return api.server.FinishTransaction(id, remote)
+}
+
+func (api *API) Transactions() (map[string]Transaction, error) {
+	return api.server.Transactions()
+}
+
+func (api *API) GetTransaction(id string, remote bool) (Transaction, error) {
+	return api.server.GetTransaction(id, remote)
+}
+
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
 	Memory           uint64 `json:"memory"`

broadcast.go

@@ -69,6 +69,7 @@ const (
 	messageTypeRecalculateCaches
 	messageTypeNodeEvent
 	messageTypeNodeStatus
+	messageTypeTransaction
 )
 
 // MarshalInternalMessage serializes the pilosa message and adds pilosa internal
@@ -116,6 +117,8 @@ func getMessage(typ byte) Message {
 		return &NodeEvent{}
 	case messageTypeNodeStatus:
 		return &NodeStatus{}
+	case messageTypeTransaction:
+		return &TransactionMessage{}
 	default:
 		panic(fmt.Sprintf("unknown message type %d", typ))
 	}
@@ -155,6 +158,8 @@ func getMessageType(m Message) byte {
 		return messageTypeNodeEvent
 	case *NodeStatus:
 		return messageTypeNodeStatus
+	case *TransactionMessage:
+		return messageTypeTransaction
 	default:
 		panic(fmt.Sprintf("don't have type for message %#v", m))
 	}

cluster.go

@@ -2589,3 +2589,15 @@ type FieldStatus struct {
 // RecalculateCaches is an internal message for recalculating all caches
 // within a holder.
 type RecalculateCaches struct{}
+
+// Transaction Actions
+const (
+	TRANSACTION_START    = "start"
+	TRANSACTION_FINISH   = "finish"
+	TRANSACTION_VALIDATE = "validate"
+)
+
+type TransactionMessage struct {
+	Transaction Transaction
+	Action      string
+}

encoding/proto/proto.go

@@ -17,6 +17,7 @@ package proto
 import (
 	"fmt"
 	"sort"
+	"time"
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/pilosa/pilosa/v2"
@@ -290,6 +291,14 @@ func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
 		}
 		decodeTranslateIDsResponse(msg, mt)
 		return nil
+	case *pilosa.TransactionMessage:
+		msg := &internal.TransactionMessage{}
+		err := proto.Unmarshal(buf, msg)
+		if err != nil {
+			return errors.Wrap(err, "unmarshaling TransactionMessage")
+		}
+		decodeTransactionMessage(msg, mt)
+		return nil
 	default:
 		panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m))
 	}
@@ -359,6 +368,8 @@ func encodeToProto(m pilosa.Message) proto.Message {
 		return encodeTranslateIDsRequest(mt)
 	case *pilosa.TranslateIDsResponse:
 		return encodeTranslateIDsResponse(mt)
+	case *pilosa.TransactionMessage:
+		return encodeTransactionMessage(mt)
 	}
 	return nil
 }
@@ -829,6 +840,35 @@ func encodeTranslateIDsResponse(response *pilosa.TranslateIDsResponse) *internal
 	}
 }
 
+func encodeTransactionMessage(msg *pilosa.TransactionMessage) *internal.TransactionMessage {
+	return &internal.TransactionMessage{
+		Action:      msg.Action,
+		Transaction: encodeTransaction(msg.Transaction),
+	}
+}
+
+func encodeTransaction(trns pilosa.Transaction) *internal.Transaction {
+	return &internal.Transaction{
+		ID:        trns.ID,
+		Active:    trns.Active,
+		Exclusive: trns.Exclusive,
+		Timeout:   int64(trns.Timeout),
+		Deadline:  encodeTransactionDeadline(trns.Deadline),
+		Stats:     encodeTransactionStats(trns.Stats),
+	}
+}
+
+func encodeTransactionDeadline(deadline time.Time) int64 {
+	if deadline.Year() > 2262 || deadline.Year() < 1678 {
+		return 0
+	}
+	return deadline.UnixNano()
+}
+
+func encodeTransactionStats(stats pilosa.TransactionStats) *internal.TransactionStats {
+	return &internal.TransactionStats{}
+}
+
 func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeInstruction) {
 	m.JobID = ri.JobID
 	m.Node = &pilosa.Node{}
@@ -1198,6 +1238,20 @@ func decodeTranslateIDsResponse(pb *internal.TranslateIDsResponse, m *pilosa.Tra
 	m.Keys = pb.Keys
 }
 
+func decodeTransactionMessage(pb *internal.TransactionMessage, m *pilosa.TransactionMessage) {
+	m.Action = pb.Action
+	decodeTransaction(pb.Transaction, &m.Transaction)
+}
+
+func decodeTransaction(pb *internal.Transaction, trns *pilosa.Transaction) {
+	trns.ID = pb.ID
+	trns.Active = pb.Active
+	trns.Exclusive = pb.Exclusive
+	trns.Timeout = time.Duration(pb.Timeout)
+	trns.Deadline = time.Unix(0, pb.Deadline)
+	// TODO m.Stats... once it has anything
+}
+
 // QueryResult types.
 const (
 	queryResultTypeNil uint32 = iota

holder.go

@@ -85,6 +85,12 @@ type Holder struct {
 	OpenTranslateStore  OpenTranslateStoreFunc
 	OpenTranslateReader OpenTranslateReaderFunc
 
+	// Func to open whatever implementation of transaction store we're using.
+	OpenTransactionStore OpenTransactionStoreFunc
+
+	// transactionManager
+	transactionManager *TransactionManager
+
 	translationSyncer translationSyncer
 
 	// Queue of fields (having a foreign index) which have
@@ -98,6 +104,22 @@ type Holder struct {
 	opening bool
 }
 
+func (h *Holder) StartTransaction(id string, timeout time.Duration, exclusive bool) (Transaction, error) {
+	return h.transactionManager.Start(id, timeout, exclusive)
+}
+
+func (h *Holder) FinishTransaction(id string) (Transaction, error) {
+	return h.transactionManager.Finish(id)
+}
+
+func (h *Holder) Transactions() (map[string]Transaction, error) {
+	return h.transactionManager.List()
+}
+
+func (h *Holder) GetTransaction(id string) (Transaction, error) {
+	return h.transactionManager.Get(id)
+}
+
 // lockedChan looks a little ridiculous admittedly, but exists for good reason.
 // The channel within is used (for example) to signal to other goroutines when
 // the Holder has finished opening (via closing the channel). However, it is
@@ -142,6 +164,8 @@ func NewHolder(partitionN int) *Holder {
 
 		OpenTranslateStore: OpenInMemTranslateStore,
 
+		OpenTransactionStore: OpenInMemTransactionStore,
+
 		translationSyncer: NopTranslationSyncer,
 
 		Logger: logger.NopLogger,
@@ -170,6 +194,13 @@ func (h *Holder) Open() error {
 		return ErrCannotOpenV1TranslateFile
 	}
 
+	if tstore, err := h.OpenTransactionStore(h.Path); err != nil {
+		return errors.Wrap(err, "opening transaction store")
+	} else {
+		h.transactionManager = NewTransactionManager(tstore)
+		h.transactionManager.Log = h.Logger
+	}
+
 	// Open path to read all index directories.
 	f, err := os.Open(h.Path)
 	if err != nil {

internal/private.pb.go

@@ -2187,6 +2187,187 @@ func (m *RecalculateCaches) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_RecalculateCaches proto.InternalMessageInfo
 
+type TransactionMessage struct {
+	Action               string       `protobuf:"bytes,1,opt,name=Action,proto3" json:"Action,omitempty"`
+	Transaction          *Transaction `protobuf:"bytes,2,opt,name=Transaction,proto3" json:"Transaction,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}     `json:"-"`
+	XXX_unrecognized     []byte       `json:"-"`
+	XXX_sizecache        int32        `json:"-"`
+}
+
+func (m *TransactionMessage) Reset()         { *m = TransactionMessage{} }
+func (m *TransactionMessage) String() string { return proto.CompactTextString(m) }
+func (*TransactionMessage) ProtoMessage()    {}
+func (*TransactionMessage) Descriptor() ([]byte, []int) {
+	return fileDescriptor_d2a91b51c7bdc125, []int{35}
+}
+func (m *TransactionMessage) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *TransactionMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_TransactionMessage.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *TransactionMessage) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_TransactionMessage.Merge(m, src)
+}
+func (m *TransactionMessage) XXX_Size() int {
+	return m.Size()
+}
+func (m *TransactionMessage) XXX_DiscardUnknown() {
+	xxx_messageInfo_TransactionMessage.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_TransactionMessage proto.InternalMessageInfo
+
+func (m *TransactionMessage) GetAction() string {
+	if m != nil {
+		return m.Action
+	}
+	return ""
+}
+
+func (m *TransactionMessage) GetTransaction() *Transaction {
+	if m != nil {
+		return m.Transaction
+	}
+	return nil
+}
+
+type Transaction struct {
+	ID                   string            `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
+	Active               bool              `protobuf:"varint,2,opt,name=Active,proto3" json:"Active,omitempty"`
+	Exclusive            bool              `protobuf:"varint,3,opt,name=Exclusive,proto3" json:"Exclusive,omitempty"`
+	Timeout              int64             `protobuf:"varint,4,opt,name=Timeout,proto3" json:"Timeout,omitempty"`
+	Deadline             int64             `protobuf:"varint,5,opt,name=Deadline,proto3" json:"Deadline,omitempty"`
+	Stats                *TransactionStats `protobuf:"bytes,6,opt,name=Stats,proto3" json:"Stats,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}          `json:"-"`
+	XXX_unrecognized     []byte            `json:"-"`
+	XXX_sizecache        int32             `json:"-"`
+}
+
+func (m *Transaction) Reset()         { *m = Transaction{} }
+func (m *Transaction) String() string { return proto.CompactTextString(m) }
+func (*Transaction) ProtoMessage()    {}
+func (*Transaction) Descriptor() ([]byte, []int) {
+	return fileDescriptor_d2a91b51c7bdc125, []int{36}
+}
+func (m *Transaction) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *Transaction) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_Transaction.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *Transaction) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Transaction.Merge(m, src)
+}
+func (m *Transaction) XXX_Size() int {
+	return m.Size()
+}
+func (m *Transaction) XXX_DiscardUnknown() {
+	xxx_messageInfo_Transaction.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Transaction proto.InternalMessageInfo
+
+func (m *Transaction) GetID() string {
+	if m != nil {
+		return m.ID
+	}
+	return ""
+}
+
+func (m *Transaction) GetActive() bool {
+	if m != nil {
+		return m.Active
+	}
+	return false
+}
+
+func (m *Transaction) GetExclusive() bool {
+	if m != nil {
+		return m.Exclusive
+	}
+	return false
+}
+
+func (m *Transaction) GetTimeout() int64 {
+	if m != nil {
+		return m.Timeout
+	}
+	return 0
+}
+
+func (m *Transaction) GetDeadline() int64 {
+	if m != nil {
+		return m.Deadline
+	}
+	return 0
+}
+
+func (m *Transaction) GetStats() *TransactionStats {
+	if m != nil {
+		return m.Stats
+	}
+	return nil
+}
+
+type TransactionStats struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *TransactionStats) Reset()         { *m = TransactionStats{} }
+func (m *TransactionStats) String() string { return proto.CompactTextString(m) }
+func (*TransactionStats) ProtoMessage()    {}
+func (*TransactionStats) Descriptor() ([]byte, []int) {
+	return fileDescriptor_d2a91b51c7bdc125, []int{37}
+}
+func (m *TransactionStats) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *TransactionStats) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_TransactionStats.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *TransactionStats) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_TransactionStats.Merge(m, src)
+}
+func (m *TransactionStats) XXX_Size() int {
+	return m.Size()
+}
+func (m *TransactionStats) XXX_DiscardUnknown() {
+	xxx_messageInfo_TransactionStats.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_TransactionStats proto.InternalMessageInfo
+
 func init() {
 	proto.RegisterType((*IndexMeta)(nil), "internal.IndexMeta")
 	proto.RegisterType((*FieldOptions)(nil), "internal.FieldOptions")
@@ -2224,94 +2405,103 @@ func init() {
 	proto.RegisterType((*UpdateCoordinatorMessage)(nil), "internal.UpdateCoordinatorMessage")
 	proto.RegisterType((*Topology)(nil), "internal.Topology")
 	proto.RegisterType((*RecalculateCaches)(nil), "internal.RecalculateCaches")
+	proto.RegisterType((*TransactionMessage)(nil), "internal.TransactionMessage")
+	proto.RegisterType((*Transaction)(nil), "internal.Transaction")
+	proto.RegisterType((*TransactionStats)(nil), "internal.TransactionStats")
 }
 
 func init() { proto.RegisterFile("private.proto", fileDescriptor_d2a91b51c7bdc125) }
 
 var fileDescriptor_d2a91b51c7bdc125 = []byte{
-	// 1298 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xcd, 0x72, 0x1b, 0xc5,
-	0x13, 0xff, 0xaf, 0x56, 0x96, 0xa5, 0x96, 0xe5, 0xc8, 0x93, 0xc4, 0xd9, 0xe4, 0x4f, 0x19, 0x31,
-	0xa4, 0x88, 0x48, 0x15, 0x26, 0x95, 0x70, 0xe0, 0x2b, 0x55, 0xc1, 0x96, 0x13, 0x44, 0xb0, 0x93,
-	0x8c, 0x9c, 0xdc, 0x38, 0x8c, 0x57, 0x53, 0xf1, 0x96, 0x57, 0xbb, 0x62, 0x77, 0xd6, 0x91, 0x73,
-	0xe0, 0x0a, 0x55, 0xbc, 0x00, 0x47, 0x1e, 0x87, 0x23, 0x8f, 0x40, 0x85, 0xa7, 0xe0, 0x46, 0x4d,
-	0xcf, 0xcc, 0x7e, 0x28, 0x16, 0x0e, 0x0e, 0xb7, 0xe9, 0x5f, 0xf7, 0xf4, 0x77, 0xf7, 0xec, 0x42,
-	0x67, 0x9a, 0x04, 0xc7, 0x5c, 0x8a, 0xcd, 0x69, 0x12, 0xcb, 0x98, 0x34, 0x83, 0x48, 0x8a, 0x24,
-	0xe2, 0xe1, 0xb5, 0x95, 0x69, 0x76, 0x10, 0x06, 0xbe, 0xc6, 0xe9, 0x03, 0x68, 0x0d, 0xa3, 0xb1,
-	0x98, 0xed, 0x0a, 0xc9, 0x09, 0x81, 0xfa, 0x43, 0x71, 0x92, 0x7a, 0x6e, 0xcf, 0xe9, 0x37, 0x19,
-	0x9e, 0xc9, 0x07, 0xb0, 0xba, 0x9f, 0x70, 0xff, 0x68, 0x67, 0x16, 0xa4, 0x52, 0x44, 0xbe, 0xf0,
-	0xea, 0xc8, 0x9d, 0x43, 0xe9, 0xaf, 0x2e, 0xac, 0xdc, 0x0f, 0x44, 0x38, 0x7e, 0x34, 0x95, 0x41,
-	0x1c, 0xa5, 0x4a, 0xd9, 0xfe, 0xc9, 0x54, 0x78, 0xcd, 0x9e, 0xd3, 0x6f, 0x31, 0x3c, 0x93, 0x77,
-	0xa0, 0xb5, 0xcd, 0xfd, 0x43, 0x81, 0x0c, 0x17, 0x19, 0x05, 0x90, 0x73, 0x47, 0xc1, 0x4b, 0x6d,
-	0xa5, 0xc3, 0x0a, 0x80, 0xf4, 0xa0, 0xbd, 0x1f, 0x4c, 0xc4, 0x93, 0x8c, 0x47, 0x32, 0x9b, 0x78,
-	0x4b, 0x78, 0xbb, 0x0c, 0x91, 0x75, 0x68, 0x3c, 0x0a, 0xc7, 0xbb, 0x41, 0xe4, 0xb5, 0x7a, 0x4e,
-	0xdf, 0x65, 0x86, 0xb2, 0x38, 0x9f, 0x79, 0x50, 0xe0, 0x7c, 0x96, 0x87, 0xdb, 0xae, 0x86, 0xbb,
-	0x17, 0x8f, 0x24, 0x8f, 0xc6, 0x3c, 0x19, 0x3f, 0x0b, 0xc4, 0x0b, 0x6f, 0x45, 0x87, 0x5b, 0x45,
-	0xd5, 0xdd, 0x2d, 0x9e, 0x0a, 0xaf, 0x83, 0x1a, 0xf1, 0x4c, 0xae, 0x41, 0x73, 0x2b, 0x90, 0x03,
-	0x31, 0x95, 0x87, 0xde, 0x6a, 0xcf, 0xe9, 0xd7, 0x59, 0x4e, 0x93, 0x4b, 0xb0, 0x34, 0xf2, 0x79,
-	0x28, 0xbc, 0x0b, 0x78, 0x41, 0x13, 0x84, 0xc2, 0xca, 0xfd, 0x38, 0x11, 0xc1, 0xf3, 0x08, 0x8b,
-	0xe0, 0x75, 0x31, 0xa8, 0x0a, 0x46, 0xde, 0x07, 0x57, 0x85, 0xb4, 0xd6, 0x73, 0xfa, 0xed, 0xdb,
-	0x6b, 0x9b, 0xb6, 0x8e, 0x9b, 0x03, 0xe1, 0x07, 0x13, 0x1e, 0x32, 0xc5, 0x45, 0x21, 0x3e, 0xf3,
-	0xc8, 0x62, 0x21, 0x3e, 0xa3, 0x14, 0x56, 0x87, 0x93, 0x69, 0x9c, 0x48, 0x26, 0xd2, 0x69, 0x1c,
-	0xa5, 0x82, 0x74, 0xc1, 0xdd, 0x49, 0x12, 0xcf, 0x41, 0xb3, 0xea, 0x48, 0x7f, 0x80, 0xee, 0x56,
-	0x18, 0xfb, 0x47, 0x03, 0x2e, 0x39, 0x13, 0xdf, 0x67, 0x22, 0x95, 0xca, 0x77, 0xed, 0x9e, 0x96,
-	0xd3, 0x84, 0x42, 0xb1, 0xde, 0x5e, 0x4d, 0xa3, 0x48, 0xa8, 0xbc, 0x60, 0xd6, 0x74, 0x79, 0xf0,
-	0x8c, 0xb1, 0x1f, 0xf2, 0x64, 0x8c, 0x35, 0xad, 0x33, 0x4d, 0x28, 0x14, 0x2d, 0x61, 0x1f, 0xd4,
-	0x99, 0x26, 0xe8, 0x10, 0xd6, 0x4a, 0xf6, 0x8d, 0x9b, 0xeb, 0xd0, 0x60, 0xf1, 0x8b, 0xe1, 0x20,
-	0xf5, 0x9c, 0x9e, 0xdb, 0xaf, 0x33, 0x43, 0x61, 0xc3, 0xc4, 0x61, 0x36, 0x89, 0x14, 0xab, 0x86,
-	0xac, 0x02, 0xa0, 0x57, 0x61, 0x09, 0xbb, 0x47, 0x45, 0x59, 0xdc, 0x55, 0x47, 0xfa, 0xa3, 0x03,
-	0xad, 0x5d, 0x3e, 0x43, 0x47, 0x52, 0x72, 0x17, 0x9a, 0xb6, 0xb6, 0x28, 0xd4, 0xbe, 0xfd, 0x5e,
-	0x91, 0xc1, 0x5c, 0x6c, 0xd3, 0xca, 0xec, 0x44, 0x32, 0x39, 0x61, 0xf9, 0x95, 0x6b, 0x5f, 0x40,
-	0xa7, 0xc2, 0x52, 0xf6, 0x8e, 0xc4, 0x89, 0xcd, 0xea, 0x91, 0x38, 0x51, 0xb1, 0x1e, 0xf3, 0x30,
-	0x13, 0x98, 0xab, 0x3a, 0xd3, 0xc4, 0xe7, 0xb5, 0x4f, 0x1d, 0xfa, 0x0c, 0xc8, 0x76, 0x22, 0xb8,
-	0x14, 0x68, 0x64, 0x57, 0xa4, 0x29, 0x7f, 0x2e, 0xce, 0xca, 0xb8, 0x5b, 0xce, 0x78, 0x9e, 0xdd,
-	0x5a, 0x29, 0xbb, 0xf4, 0x26, 0x90, 0x81, 0x08, 0x85, 0x14, 0x66, 0xba, 0xff, 0x41, 0x2f, 0x1d,
-	0x59, 0x1f, 0xce, 0x96, 0x25, 0x37, 0xa0, 0xae, 0x56, 0x05, 0x1a, 0x6b, 0xdf, 0xbe, 0x58, 0xe4,
-	0x29, 0xdf, 0x22, 0x0c, 0x05, 0x68, 0x68, 0x95, 0xa2, 0x97, 0x6f, 0x18, 0x58, 0xa5, 0x95, 0x6e,
-	0x1a, 0x53, 0x2e, 0x9a, 0x5a, 0x2f, 0x4c, 0x95, 0xd7, 0x8c, 0xb1, 0x76, 0xcf, 0x86, 0x7b, 0x5e,
-	0x6b, 0xd4, 0x87, 0xff, 0x6b, 0x0d, 0x5f, 0x1d, 0xf3, 0x20, 0xe4, 0x07, 0xe1, 0xbf, 0xaa, 0x48,
-	0xc5, 0x71, 0x0f, 0x96, 0xf1, 0xee, 0x70, 0x60, 0x7a, 0xdb, 0x92, 0xf4, 0x3b, 0x28, 0xc6, 0x64,
-	0x8f, 0x4f, 0x84, 0xd1, 0x86, 0xe7, 0x3c, 0xde, 0xda, 0xd9, 0xf1, 0x2a, 0xc3, 0x6a, 0xb4, 0xd4,
-	0xaa, 0x76, 0x95, 0x61, 0x24, 0xe8, 0x1d, 0x68, 0x8c, 0xfc, 0x43, 0x31, 0xe1, 0xe4, 0x43, 0x58,
-	0x46, 0x0f, 0x45, 0x6a, 0x3a, 0xfa, 0xc2, 0x5c, 0xa5, 0x98, 0xe5, 0xd3, 0xd4, 0x44, 0x76, 0xaa,
-	0x4f, 0x1f, 0xc1, 0xb2, 0x31, 0x8c, 0x13, 0xbd, 0xa0, 0xe2, 0x56, 0x86, 0xdc, 0x80, 0x06, 0x3a,
-	0x9b, 0x7a, 0xf5, 0x79, 0xab, 0x88, 0x33, 0xc3, 0xa6, 0x3b, 0xe0, 0x3e, 0x65, 0x43, 0x35, 0xd8,
-	0xe8, 0xb0, 0x35, 0x6a, 0x28, 0xe5, 0xca, 0xd7, 0x71, 0x2a, 0x4d, 0x5a, 0xf1, 0xac, 0xb0, 0xc7,
-	0x71, 0x22, 0x31, 0xa5, 0x1d, 0x86, 0x67, 0x9a, 0x42, 0x7d, 0x2f, 0x1e, 0x0b, 0xb2, 0x0a, 0xb5,
-	0xe1, 0xc0, 0xe8, 0xa8, 0x0d, 0x07, 0xe4, 0x5d, 0x54, 0x6f, 0x32, 0xd9, 0x29, 0x9c, 0x78, 0xca,
-	0x86, 0x0c, 0x0d, 0x5f, 0x87, 0xce, 0x30, 0xdd, 0x8e, 0xe3, 0x64, 0x1c, 0x44, 0x5c, 0xc6, 0x89,
-	0x79, 0xf2, 0xaa, 0x20, 0x8e, 0x96, 0xe4, 0x52, 0x3f, 0x46, 0x2d, 0xa6, 0x09, 0x7a, 0x0f, 0xba,
-	0xca, 0x28, 0x12, 0xb6, 0x3d, 0xd6, 0xa1, 0xa1, 0xb0, 0xdc, 0x09, 0x43, 0x15, 0x1a, 0x6a, 0x65,
-	0x0d, 0xdf, 0x6a, 0x0d, 0x3b, 0xc7, 0x22, 0x92, 0xa5, 0x06, 0x43, 0x1a, 0x15, 0x74, 0x98, 0x26,
-	0x08, 0xd5, 0x01, 0x9a, 0x48, 0x56, 0x8b, 0x48, 0x14, 0xca, 0x90, 0x47, 0x7f, 0x76, 0x00, 0xac,
-	0x43, 0x59, 0x9a, 0x5f, 0x71, 0x16, 0x5f, 0x21, 0x7d, 0xdb, 0x28, 0x66, 0xb8, 0xba, 0x85, 0x94,
-	0xc6, 0x99, 0x6d, 0xa4, 0x8f, 0x8b, 0x46, 0xd2, 0x25, 0xbd, 0x3c, 0xd7, 0x00, 0xda, 0x6a, 0xd1,
-	0x4e, 0x8f, 0xa1, 0x5d, 0xc2, 0x17, 0x34, 0x95, 0xed, 0x92, 0xda, 0xbc, 0x4a, 0xc4, 0x8d, 0x4a,
-	0xdb, 0x2b, 0x0f, 0xa1, 0x5d, 0x82, 0x4f, 0xd5, 0xd8, 0x87, 0x0b, 0xd5, 0xb1, 0xb5, 0xcf, 0xc1,
-	0x3c, 0x4c, 0x03, 0xe8, 0x6c, 0x87, 0x59, 0x2a, 0x45, 0x62, 0xd4, 0xa9, 0x37, 0x44, 0x03, 0x79,
-	0xf1, 0x0a, 0xe0, 0xf4, 0xfa, 0x91, 0xeb, 0xb0, 0xa4, 0xd2, 0xa8, 0xa7, 0xef, 0xf5, 0x1c, 0x6b,
-	0x26, 0x7d, 0x06, 0xcd, 0xad, 0xd1, 0xf0, 0x41, 0x12, 0x67, 0xd3, 0x53, 0x9d, 0xb6, 0x1f, 0x48,
-	0xb5, 0xd2, 0x07, 0x52, 0x57, 0x3f, 0xf6, 0x2e, 0x7e, 0x24, 0xe0, 0xcb, 0xde, 0xd5, 0x2f, 0x7b,
-	0xdd, 0x20, 0x5c, 0xad, 0xeb, 0x35, 0xbd, 0x59, 0xd5, 0xd0, 0x9f, 0x67, 0x3f, 0xd9, 0x37, 0xda,
-	0x2d, 0xde, 0x68, 0xa5, 0x54, 0xaf, 0xbf, 0xff, 0x52, 0xe9, 0x5f, 0x35, 0x58, 0x63, 0x22, 0x0d,
-	0x5e, 0x8a, 0x61, 0x94, 0xca, 0x24, 0xf3, 0xd5, 0x96, 0x50, 0xf7, 0xbf, 0x89, 0x0f, 0x4c, 0xb6,
-	0x5d, 0xa6, 0x89, 0x37, 0xe9, 0x74, 0x72, 0x0b, 0xda, 0xf3, 0x33, 0xfb, 0xba, 0x68, 0x59, 0x84,
-	0xdc, 0x82, 0xe5, 0x51, 0x9c, 0x25, 0x7e, 0xde, 0xbe, 0xa5, 0xb5, 0xaa, 0x3d, 0xd3, 0x6c, 0x66,
-	0xc5, 0xc8, 0x13, 0x20, 0xfb, 0x09, 0x8f, 0xd2, 0x90, 0x2b, 0x67, 0xed, 0xe5, 0xe6, 0xfc, 0x67,
-	0x41, 0x49, 0xa6, 0xa2, 0xe7, 0x94, 0xcb, 0xe4, 0x93, 0xf2, 0x7c, 0x7a, 0xcb, 0xe8, 0xf5, 0xa5,
-	0xaa, 0xd7, 0xa6, 0xe5, 0xcb, 0x73, 0x7c, 0x77, 0xae, 0x53, 0xbd, 0x06, 0x5e, 0xbc, 0x52, 0x5c,
-	0xac, 0xb0, 0x59, 0x55, 0x9a, 0xfe, 0xe4, 0xc0, 0x4a, 0xd9, 0xb3, 0x37, 0xda, 0x0b, 0x79, 0xc1,
-	0x6b, 0x67, 0x7f, 0x77, 0xd8, 0x82, 0xd7, 0x4f, 0xfb, 0xd2, 0x5b, 0x2a, 0x7f, 0x8b, 0x64, 0x70,
-	0x65, 0x41, 0xba, 0xde, 0xc2, 0xa9, 0x1e, 0xb4, 0x1f, 0xf3, 0x44, 0x06, 0x4a, 0xa5, 0x79, 0x68,
-	0x97, 0x58, 0x19, 0xa2, 0x47, 0x70, 0xf5, 0xb5, 0xe6, 0xdb, 0x8e, 0x27, 0x53, 0xd5, 0xe5, 0x6f,
-	0xd1, 0x84, 0x6a, 0x51, 0x27, 0x89, 0x69, 0xbf, 0x16, 0xd3, 0x04, 0xfd, 0x0c, 0x2e, 0x8f, 0x84,
-	0x2c, 0xb5, 0x9e, 0x9d, 0xa1, 0x1e, 0xb8, 0x7b, 0xe2, 0xc5, 0x82, 0x00, 0x15, 0x8b, 0x7e, 0x09,
-	0xde, 0xd3, 0xe9, 0x98, 0x4b, 0x71, 0xae, 0xdb, 0x5b, 0xd0, 0xdc, 0x8f, 0xa7, 0x71, 0x18, 0x3f,
-	0x3f, 0x39, 0x63, 0x97, 0x79, 0xb0, 0xac, 0x5f, 0x25, 0xbd, 0x1c, 0x5b, 0xcc, 0x92, 0xf4, 0xa2,
-	0x1a, 0x53, 0x9f, 0x87, 0x7e, 0x16, 0x2a, 0x37, 0xd4, 0x47, 0x73, 0xba, 0xd5, 0xfd, 0xed, 0xd5,
-	0x86, 0xf3, 0xfb, 0xab, 0x0d, 0xe7, 0x8f, 0x57, 0x1b, 0xce, 0x2f, 0x7f, 0x6e, 0xfc, 0xef, 0xa0,
-	0x81, 0xbf, 0x8c, 0x77, 0xfe, 0x0e, 0x00, 0x00, 0xff, 0xff, 0xa0, 0xbb, 0xce, 0xd0, 0x5b, 0x0e,
-	0x00, 0x00,
+	// 1395 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x17, 0xcd, 0x72, 0xdb, 0x44,
+	0x18, 0x59, 0x8e, 0x63, 0x7f, 0x8e, 0x53, 0x67, 0xdb, 0xa6, 0x6a, 0x60, 0x82, 0x59, 0x3a, 0xd4,
+	0x74, 0x86, 0xd0, 0x69, 0x99, 0xe1, 0xb7, 0x33, 0x6d, 0xe2, 0xb4, 0x98, 0x92, 0xb4, 0x5d, 0xa7,
+	0xbd, 0x71, 0xd8, 0xc8, 0x3b, 0x8d, 0x26, 0xb2, 0x64, 0xa4, 0x55, 0xea, 0xf4, 0xc0, 0x15, 0x66,
+	0x78, 0x01, 0x8e, 0xbc, 0x07, 0x2f, 0xc0, 0x91, 0x47, 0x60, 0xca, 0x53, 0x70, 0x63, 0xf6, 0xdb,
+	0x5d, 0x49, 0x76, 0x1c, 0x52, 0x52, 0x6e, 0xfb, 0xfd, 0xff, 0x7f, 0x9f, 0x04, 0xad, 0x71, 0x12,
+	0x1c, 0x71, 0x29, 0x36, 0xc6, 0x49, 0x2c, 0x63, 0x52, 0x0f, 0x22, 0x29, 0x92, 0x88, 0x87, 0x6b,
+	0x4b, 0xe3, 0x6c, 0x3f, 0x0c, 0x7c, 0x8d, 0xa7, 0x0f, 0xa0, 0xd1, 0x8f, 0x86, 0x62, 0xb2, 0x23,
+	0x24, 0x27, 0x04, 0xaa, 0x0f, 0xc5, 0x71, 0xea, 0xb9, 0x1d, 0xa7, 0x5b, 0x67, 0xf8, 0x26, 0x1f,
+	0xc0, 0xf2, 0x5e, 0xc2, 0xfd, 0xc3, 0xed, 0x49, 0x90, 0x4a, 0x11, 0xf9, 0xc2, 0xab, 0x22, 0x75,
+	0x06, 0x4b, 0x7f, 0x75, 0x61, 0xe9, 0x7e, 0x20, 0xc2, 0xe1, 0xa3, 0xb1, 0x0c, 0xe2, 0x28, 0x55,
+	0xca, 0xf6, 0x8e, 0xc7, 0xc2, 0xab, 0x77, 0x9c, 0x6e, 0x83, 0xe1, 0x9b, 0xbc, 0x03, 0x8d, 0x2d,
+	0xee, 0x1f, 0x08, 0x24, 0xb8, 0x48, 0x28, 0x10, 0x39, 0x75, 0x10, 0xbc, 0xd4, 0x56, 0x5a, 0xac,
+	0x40, 0x90, 0x0e, 0x34, 0xf7, 0x82, 0x91, 0x78, 0x92, 0xf1, 0x48, 0x66, 0x23, 0x6f, 0x01, 0xa5,
+	0xcb, 0x28, 0xb2, 0x0a, 0xb5, 0x47, 0xe1, 0x70, 0x27, 0x88, 0xbc, 0x46, 0xc7, 0xe9, 0xba, 0xcc,
+	0x40, 0x16, 0xcf, 0x27, 0x1e, 0x14, 0x78, 0x3e, 0xc9, 0xc3, 0x6d, 0x4e, 0x87, 0xbb, 0x1b, 0x0f,
+	0x24, 0x8f, 0x86, 0x3c, 0x19, 0x3e, 0x0b, 0xc4, 0x0b, 0x6f, 0x49, 0x87, 0x3b, 0x8d, 0x55, 0xb2,
+	0x9b, 0x3c, 0x15, 0x5e, 0x0b, 0x35, 0xe2, 0x9b, 0xac, 0x41, 0x7d, 0x33, 0x90, 0x3d, 0x31, 0x96,
+	0x07, 0xde, 0x72, 0xc7, 0xe9, 0x56, 0x59, 0x0e, 0x93, 0x4b, 0xb0, 0x30, 0xf0, 0x79, 0x28, 0xbc,
+	0x0b, 0x28, 0xa0, 0x01, 0x42, 0x61, 0xe9, 0x7e, 0x9c, 0x88, 0xe0, 0x79, 0x84, 0x45, 0xf0, 0xda,
+	0x18, 0xd4, 0x14, 0x8e, 0xbc, 0x0f, 0xae, 0x0a, 0x69, 0xa5, 0xe3, 0x74, 0x9b, 0xb7, 0x56, 0x36,
+	0x6c, 0x1d, 0x37, 0x7a, 0xc2, 0x0f, 0x46, 0x3c, 0x64, 0x8a, 0x8a, 0x4c, 0x7c, 0xe2, 0x91, 0xd3,
+	0x99, 0xf8, 0x84, 0x52, 0x58, 0xee, 0x8f, 0xc6, 0x71, 0x22, 0x99, 0x48, 0xc7, 0x71, 0x94, 0x0a,
+	0xd2, 0x06, 0x77, 0x3b, 0x49, 0x3c, 0x07, 0xcd, 0xaa, 0x27, 0xfd, 0x01, 0xda, 0x9b, 0x61, 0xec,
+	0x1f, 0xf6, 0xb8, 0xe4, 0x4c, 0x7c, 0x9f, 0x89, 0x54, 0x2a, 0xdf, 0xb5, 0x7b, 0x9a, 0x4f, 0x03,
+	0x0a, 0x8b, 0xf5, 0xf6, 0x2a, 0x1a, 0x8b, 0x80, 0xca, 0x0b, 0x66, 0x4d, 0x97, 0x07, 0xdf, 0x18,
+	0xfb, 0x01, 0x4f, 0x86, 0x58, 0xd3, 0x2a, 0xd3, 0x80, 0xc2, 0xa2, 0x25, 0xec, 0x83, 0x2a, 0xd3,
+	0x00, 0xed, 0xc3, 0x4a, 0xc9, 0xbe, 0x71, 0x73, 0x15, 0x6a, 0x2c, 0x7e, 0xd1, 0xef, 0xa5, 0x9e,
+	0xd3, 0x71, 0xbb, 0x55, 0x66, 0x20, 0x6c, 0x98, 0x38, 0xcc, 0x46, 0x91, 0x22, 0x55, 0x90, 0x54,
+	0x20, 0xe8, 0x55, 0x58, 0xc0, 0xee, 0x51, 0x51, 0x16, 0xb2, 0xea, 0x49, 0x7f, 0x74, 0xa0, 0xb1,
+	0xc3, 0x27, 0xe8, 0x48, 0x4a, 0xee, 0x40, 0xdd, 0xd6, 0x16, 0x99, 0x9a, 0xb7, 0xde, 0x2b, 0x32,
+	0x98, 0xb3, 0x6d, 0x58, 0x9e, 0xed, 0x48, 0x26, 0xc7, 0x2c, 0x17, 0x59, 0xfb, 0x12, 0x5a, 0x53,
+	0x24, 0x65, 0xef, 0x50, 0x1c, 0xdb, 0xac, 0x1e, 0x8a, 0x63, 0x15, 0xeb, 0x11, 0x0f, 0x33, 0x81,
+	0xb9, 0xaa, 0x32, 0x0d, 0x7c, 0x51, 0xf9, 0xcc, 0xa1, 0xcf, 0x80, 0x6c, 0x25, 0x82, 0x4b, 0x81,
+	0x46, 0x76, 0x44, 0x9a, 0xf2, 0xe7, 0xe2, 0xac, 0x8c, 0xbb, 0xe5, 0x8c, 0xe7, 0xd9, 0xad, 0x94,
+	0xb2, 0x4b, 0x6f, 0x00, 0xe9, 0x89, 0x50, 0x48, 0x61, 0xa6, 0xfb, 0x5f, 0xf4, 0xd2, 0x81, 0xf5,
+	0xe1, 0x6c, 0x5e, 0x72, 0x1d, 0xaa, 0x6a, 0x55, 0xa0, 0xb1, 0xe6, 0xad, 0x8b, 0x45, 0x9e, 0xf2,
+	0x2d, 0xc2, 0x90, 0x81, 0x86, 0x56, 0x29, 0x7a, 0xf9, 0x9a, 0x81, 0x4d, 0xb5, 0xd2, 0x0d, 0x63,
+	0xca, 0x45, 0x53, 0xab, 0x85, 0xa9, 0xf2, 0x9a, 0x31, 0xd6, 0xee, 0xda, 0x70, 0xcf, 0x6b, 0x8d,
+	0xfa, 0xf0, 0xb6, 0xd6, 0x70, 0xef, 0x88, 0x07, 0x21, 0xdf, 0x0f, 0xff, 0x53, 0x45, 0xa6, 0x1c,
+	0xf7, 0x60, 0x11, 0x65, 0xfb, 0x3d, 0xd3, 0xdb, 0x16, 0xa4, 0xdf, 0x41, 0x31, 0x26, 0xbb, 0x7c,
+	0x24, 0x8c, 0x36, 0x7c, 0xe7, 0xf1, 0x56, 0xce, 0x8e, 0x57, 0x19, 0x56, 0xa3, 0xa5, 0x56, 0xb5,
+	0xab, 0x0c, 0x23, 0x40, 0x6f, 0x43, 0x6d, 0xe0, 0x1f, 0x88, 0x11, 0x27, 0x1f, 0xc2, 0x22, 0x7a,
+	0x28, 0x52, 0xd3, 0xd1, 0x17, 0x66, 0x2a, 0xc5, 0x2c, 0x9d, 0xa6, 0x26, 0xb2, 0xb9, 0x3e, 0x7d,
+	0x04, 0x8b, 0xc6, 0x30, 0x4e, 0xf4, 0x29, 0x15, 0xb7, 0x3c, 0xe4, 0x3a, 0xd4, 0xd0, 0xd9, 0xd4,
+	0xab, 0xce, 0x5a, 0x45, 0x3c, 0x33, 0x64, 0xba, 0x0d, 0xee, 0x53, 0xd6, 0x57, 0x83, 0x8d, 0x0e,
+	0x5b, 0xa3, 0x06, 0x52, 0xae, 0x7c, 0x1d, 0xa7, 0xd2, 0xa4, 0x15, 0xdf, 0x0a, 0xf7, 0x38, 0x4e,
+	0x24, 0xa6, 0xb4, 0xc5, 0xf0, 0x4d, 0x53, 0xa8, 0xee, 0xc6, 0x43, 0x41, 0x96, 0xa1, 0xd2, 0xef,
+	0x19, 0x1d, 0x95, 0x7e, 0x8f, 0xbc, 0x8b, 0xea, 0x4d, 0x26, 0x5b, 0x85, 0x13, 0x4f, 0x59, 0x9f,
+	0xa1, 0xe1, 0x6b, 0xd0, 0xea, 0xa7, 0x5b, 0x71, 0x9c, 0x0c, 0x83, 0x88, 0xcb, 0x38, 0x31, 0x27,
+	0x6f, 0x1a, 0x89, 0xa3, 0x25, 0xb9, 0xd4, 0xc7, 0xa8, 0xc1, 0x34, 0x40, 0xef, 0x42, 0x5b, 0x19,
+	0x45, 0xc0, 0xb6, 0xc7, 0x2a, 0xd4, 0x14, 0x2e, 0x77, 0xc2, 0x40, 0x85, 0x86, 0x4a, 0x59, 0xc3,
+	0xb7, 0x5a, 0xc3, 0xf6, 0x91, 0x88, 0x64, 0xa9, 0xc1, 0x10, 0x46, 0x05, 0x2d, 0xa6, 0x01, 0x42,
+	0x75, 0x80, 0x26, 0x92, 0xe5, 0x22, 0x12, 0x85, 0x65, 0x48, 0xa3, 0x3f, 0x3b, 0x00, 0xd6, 0xa1,
+	0x2c, 0xcd, 0x45, 0x9c, 0xd3, 0x45, 0x48, 0xd7, 0x36, 0x8a, 0x19, 0xae, 0x76, 0xc1, 0xa5, 0xf1,
+	0xcc, 0x36, 0xd2, 0xc7, 0x45, 0x23, 0xe9, 0x92, 0x5e, 0x9e, 0x69, 0x00, 0x6d, 0xb5, 0x68, 0xa7,
+	0xc7, 0xd0, 0x2c, 0xe1, 0x4f, 0x69, 0x2a, 0xdb, 0x25, 0x95, 0x59, 0x95, 0x88, 0x37, 0x2a, 0x6d,
+	0xaf, 0x3c, 0x84, 0x66, 0x09, 0x3d, 0x57, 0x63, 0x17, 0x2e, 0x4c, 0x8f, 0xad, 0x3d, 0x07, 0xb3,
+	0x68, 0x1a, 0x40, 0x6b, 0x2b, 0xcc, 0x52, 0x29, 0x12, 0xa3, 0x4e, 0xdd, 0x10, 0x8d, 0xc8, 0x8b,
+	0x57, 0x20, 0xe6, 0xd7, 0x8f, 0x5c, 0x83, 0x05, 0x95, 0x46, 0x3d, 0x7d, 0x27, 0x73, 0xac, 0x89,
+	0xf4, 0x19, 0xd4, 0x37, 0x07, 0xfd, 0x07, 0x49, 0x9c, 0x8d, 0xe7, 0x3a, 0x6d, 0x3f, 0x90, 0x2a,
+	0xa5, 0x0f, 0xa4, 0xb6, 0x3e, 0xf6, 0x2e, 0x7e, 0x24, 0xe0, 0x65, 0x6f, 0xeb, 0xcb, 0x5e, 0x35,
+	0x18, 0xae, 0xd6, 0xf5, 0x8a, 0xde, 0xac, 0x6a, 0xe8, 0xcf, 0xb3, 0x9f, 0xec, 0x8d, 0x76, 0x8b,
+	0x1b, 0xad, 0x94, 0xea, 0xf5, 0xf7, 0x7f, 0x2a, 0xfd, 0xbb, 0x02, 0x2b, 0x4c, 0xa4, 0xc1, 0x4b,
+	0xd1, 0x8f, 0x52, 0x99, 0x64, 0xbe, 0xda, 0x12, 0x4a, 0xfe, 0x9b, 0x78, 0xdf, 0x64, 0xdb, 0x65,
+	0x1a, 0x78, 0x9d, 0x4e, 0x27, 0x37, 0xa1, 0x39, 0x3b, 0xb3, 0x27, 0x59, 0xcb, 0x2c, 0xe4, 0x26,
+	0x2c, 0x0e, 0xe2, 0x2c, 0xf1, 0xf3, 0xf6, 0x2d, 0xad, 0x55, 0xed, 0x99, 0x26, 0x33, 0xcb, 0x46,
+	0x9e, 0x00, 0xd9, 0x4b, 0x78, 0x94, 0x86, 0x5c, 0x39, 0x6b, 0x85, 0xeb, 0xb3, 0x9f, 0x05, 0x25,
+	0x9e, 0x29, 0x3d, 0x73, 0x84, 0xc9, 0x27, 0xe5, 0xf9, 0xf4, 0x16, 0xd1, 0xeb, 0x4b, 0xd3, 0x5e,
+	0x9b, 0x96, 0x2f, 0xcf, 0xf1, 0x9d, 0x99, 0x4e, 0xf5, 0x6a, 0x28, 0x78, 0xa5, 0x10, 0x9c, 0x22,
+	0xb3, 0x69, 0x6e, 0xfa, 0x93, 0x03, 0x4b, 0x65, 0xcf, 0x5e, 0x6b, 0x2f, 0xe4, 0x05, 0xaf, 0x9c,
+	0xfd, 0xdd, 0x61, 0x0b, 0x5e, 0x9d, 0xf7, 0xa5, 0xb7, 0x50, 0xfe, 0x16, 0xc9, 0xe0, 0xca, 0x29,
+	0xe9, 0x7a, 0x03, 0xa7, 0x3a, 0xd0, 0x7c, 0xcc, 0x13, 0x19, 0x28, 0x95, 0xe6, 0xd0, 0x2e, 0xb0,
+	0x32, 0x8a, 0x1e, 0xc2, 0xd5, 0x13, 0xcd, 0xb7, 0x15, 0x8f, 0xc6, 0xaa, 0xcb, 0xdf, 0xa0, 0x09,
+	0xd5, 0xa2, 0x4e, 0x12, 0xd3, 0x7e, 0x0d, 0xa6, 0x01, 0xfa, 0x39, 0x5c, 0x1e, 0x08, 0x59, 0x6a,
+	0x3d, 0x3b, 0x43, 0x1d, 0x70, 0x77, 0xc5, 0x8b, 0x53, 0x02, 0x54, 0x24, 0xfa, 0x15, 0x78, 0x4f,
+	0xc7, 0x43, 0x2e, 0xc5, 0xb9, 0xa4, 0x37, 0xa1, 0xbe, 0x17, 0x8f, 0xe3, 0x30, 0x7e, 0x7e, 0x7c,
+	0xc6, 0x2e, 0xf3, 0x60, 0x51, 0x5f, 0x25, 0xbd, 0x1c, 0x1b, 0xcc, 0x82, 0xf4, 0xa2, 0x1a, 0x53,
+	0x9f, 0x87, 0x7e, 0x16, 0x2a, 0x37, 0xd4, 0x47, 0x73, 0x4a, 0x85, 0x19, 0x04, 0x8e, 0x89, 0x2b,
+	0x1d, 0xba, 0x7b, 0x88, 0xb0, 0x87, 0x4e, 0x43, 0xe4, 0x53, 0x68, 0x96, 0xb8, 0x4d, 0x02, 0x2f,
+	0xcf, 0xcc, 0x8b, 0x26, 0xb2, 0x32, 0x27, 0xfd, 0xcd, 0x99, 0x92, 0x3c, 0x71, 0xca, 0x8d, 0xc1,
+	0x23, 0x5d, 0x94, 0x3a, 0x33, 0x90, 0x8a, 0x75, 0x7b, 0xe2, 0x87, 0x59, 0xaa, 0x48, 0xfa, 0x7a,
+	0x17, 0x08, 0x15, 0xab, 0xfa, 0x33, 0x8c, 0x33, 0x69, 0x36, 0xa7, 0x05, 0xd5, 0x4f, 0x5a, 0x4f,
+	0xf0, 0x61, 0x18, 0x44, 0x02, 0xbb, 0xd4, 0x65, 0x39, 0x4c, 0x6e, 0xea, 0x6d, 0x6f, 0x47, 0x6d,
+	0x6d, 0xae, 0xfb, 0xc8, 0xa1, 0x2f, 0x41, 0x4a, 0x09, 0xb4, 0x67, 0x49, 0x9b, 0xed, 0xdf, 0x5f,
+	0xad, 0x3b, 0x7f, 0xbc, 0x5a, 0x77, 0xfe, 0x7c, 0xb5, 0xee, 0xfc, 0xf2, 0xd7, 0xfa, 0x5b, 0xfb,
+	0x35, 0xfc, 0xd7, 0xbe, 0xfd, 0x4f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x2f, 0x76, 0xf5, 0x59, 0x94,
+	0x0f, 0x00, 0x00,
 }
 
 func (m *IndexMeta) Marshal() (dAtA []byte, err error) {
@@ -4146,6 +4336,155 @@ func (m *RecalculateCaches) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *TransactionMessage) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *TransactionMessage) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *TransactionMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.Transaction != nil {
+		{
+			size, err := m.Transaction.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintPrivate(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x12
+	}
+	if len(m.Action) > 0 {
+		i -= len(m.Action)
+		copy(dAtA[i:], m.Action)
+		i = encodeVarintPrivate(dAtA, i, uint64(len(m.Action)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *Transaction) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *Transaction) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *Transaction) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.Stats != nil {
+		{
+			size, err := m.Stats.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintPrivate(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x32
+	}
+	if m.Deadline != 0 {
+		i = encodeVarintPrivate(dAtA, i, uint64(m.Deadline))
+		i--
+		dAtA[i] = 0x28
+	}
+	if m.Timeout != 0 {
+		i = encodeVarintPrivate(dAtA, i, uint64(m.Timeout))
+		i--
+		dAtA[i] = 0x20
+	}
+	if m.Exclusive {
+		i--
+		if m.Exclusive {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x18
+	}
+	if m.Active {
+		i--
+		if m.Active {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x10
+	}
+	if len(m.ID) > 0 {
+		i -= len(m.ID)
+		copy(dAtA[i:], m.ID)
+		i = encodeVarintPrivate(dAtA, i, uint64(len(m.ID)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *TransactionStats) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *TransactionStats) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *TransactionStats) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return len(dAtA) - i, nil
+}
+
 func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int {
 	offset -= sovPrivate(v)
 	base := offset
@@ -4988,6 +5327,70 @@ func (m *RecalculateCaches) Size() (n int) {
 	return n
 }
 
+func (m *TransactionMessage) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.Action)
+	if l > 0 {
+		n += 1 + l + sovPrivate(uint64(l))
+	}
+	if m.Transaction != nil {
+		l = m.Transaction.Size()
+		n += 1 + l + sovPrivate(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *Transaction) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.ID)
+	if l > 0 {
+		n += 1 + l + sovPrivate(uint64(l))
+	}
+	if m.Active {
+		n += 2
+	}
+	if m.Exclusive {
+		n += 2
+	}
+	if m.Timeout != 0 {
+		n += 1 + sovPrivate(uint64(m.Timeout))
+	}
+	if m.Deadline != 0 {
+		n += 1 + sovPrivate(uint64(m.Deadline))
+	}
+	if m.Stats != nil {
+		l = m.Stats.Size()
+		n += 1 + l + sovPrivate(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *TransactionStats) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func sovPrivate(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
@@ -10137,6 +10540,382 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *TransactionMessage) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowPrivate
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: TransactionMessage: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: TransactionMessage: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Action", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Action = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Transaction", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Transaction == nil {
+				m.Transaction = &Transaction{}
+			}
+			if err := m.Transaction.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipPrivate(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *Transaction) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowPrivate
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: Transaction: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: Transaction: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ID", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ID = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Active", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.Active = bool(v != 0)
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Exclusive", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.Exclusive = bool(v != 0)
+		case 4:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Timeout", wireType)
+			}
+			m.Timeout = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Timeout |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 5:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Deadline", wireType)
+			}
+			m.Deadline = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Deadline |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 6:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Stats", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPrivate
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Stats == nil {
+				m.Stats = &TransactionStats{}
+			}
+			if err := m.Stats.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipPrivate(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *TransactionStats) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowPrivate
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: TransactionStats: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: TransactionStats: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		default:
+			iNdEx = preIndex
+			skippy, err := skipPrivate(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthPrivate
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func skipPrivate(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0

internal/private.proto

@@ -207,3 +207,19 @@ message Topology {
 }
 
 message RecalculateCaches {}
+
+message TransactionMessage {
+	string Action = 1;
+	Transaction Transaction = 2;
+}
+
+message Transaction {
+	string ID = 1;
+	bool Active = 2;
+	bool Exclusive = 3;
+	int64 Timeout = 4;
+	int64 Deadline = 5;
+	TransactionStats Stats = 6;
+}
+
+message TransactionStats {}

server.go

@@ -28,6 +28,8 @@ import (
 	"time"
 
 	"github.com/molecula/ext"
+	uuid "github.com/satori/go.uuid"
+
 	// extensions pulls in some extensions depending on build tags
 	_ "github.com/pilosa/pilosa/v2/extensions"
 	"github.com/pilosa/pilosa/v2/logger"
@@ -794,11 +796,42 @@ func (s *Server) receiveMessage(m Message) error {
 		}
 	case *NodeStatus:
 		s.handleRemoteStatus(obj)
+	case *TransactionMessage:
+		err := s.handleTransactionMessage(obj)
+		if err != nil {
+			return errors.Wrapf(err, "handling transaction message: %v", obj)
+		}
 	}
 
 	return nil
 }
 
+func (s *Server) handleTransactionMessage(tm *TransactionMessage) error {
+	mtrns := tm.Transaction // message transaction
+	switch tm.Action {
+	case TRANSACTION_START:
+		_, err := s.StartTransaction(mtrns.ID, mtrns.Timeout, mtrns.Exclusive, true)
+		if err != nil {
+			return errors.Wrap(err, "starting transaction locally")
+		}
+	case TRANSACTION_FINISH:
+		_, err := s.FinishTransaction(mtrns.ID, true)
+		if err != nil {
+			return errors.Wrap(err, "finishing transaction locally")
+		}
+	case TRANSACTION_VALIDATE:
+		trns, err := s.GetTransaction(mtrns.ID, true)
+		if err != nil {
+			return errors.Wrap(err, "getting local transaction to validate")
+		}
+		err = CompareTransactions(mtrns, trns)
+		return errors.Wrap(err, "comparing transactions")
+	default:
+		return errors.Errorf("unknown transaction action: '%s'", tm.Action)
+	}
+	return nil
+}
+
 // SendSync represents an implementation of Broadcaster.
 func (s *Server) SendSync(m Message) error {
 	var eg errgroup.Group
@@ -991,6 +1024,125 @@ func (s *Server) monitorRuntime() {
 	}
 }
 
+func (srv *Server) StartTransaction(id string, timeout time.Duration, exclusive bool, remote bool) (Transaction, error) {
+	node := srv.node()
+	if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
+		return Transaction{}, ErrNodeNotCoordinator
+	}
+	if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) {
+		return Transaction{}, errors.New("got a remote start call to coordinator or single node cluster... shouldn't ever happen")
+	}
+	// empty string id should generate an id
+
+	if !remote { // we are the coordinator,
+		if id == "" {
+			id = uuid.NewV4().String()
+		}
+		trns, err := srv.holder.StartTransaction(id, timeout, exclusive)
+		if err != nil {
+			return trns, errors.Wrap(err, "starting transaction")
+		}
+		err = srv.SendSync(
+			&TransactionMessage{
+				Action:      TRANSACTION_START,
+				Transaction: trns,
+			})
+		if err != nil {
+			// try to clean up, but ignore errors
+			srv.holder.FinishTransaction(id)
+			srv.SendSync(
+				&TransactionMessage{
+					Action:      TRANSACTION_FINISH,
+					Transaction: trns,
+				},
+			)
+			return trns, errors.Wrap(err, "broadcasting transaction start")
+		}
+		return trns, nil
+	} else { // remote
+		return srv.holder.StartTransaction(id, timeout, exclusive)
+	}
+
+}
+
+func (srv *Server) FinishTransaction(id string, remote bool) (Transaction, error) {
+	node := srv.node()
+	if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
+		return Transaction{}, ErrNodeNotCoordinator
+	}
+	if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) {
+		return Transaction{}, errors.New("got a remote finish call to coordinator or single node cluster... shouldn't ever happen")
+	}
+
+	if !remote {
+		trns, err := srv.holder.FinishTransaction(id)
+		if err != nil {
+			return trns, errors.Wrap(err, "finishing transaction")
+		}
+		err = srv.SendSync(
+			&TransactionMessage{
+				Action:      TRANSACTION_FINISH,
+				Transaction: trns,
+			},
+		)
+		if err != nil {
+			srv.logger.Printf("error broadcasting transaction finish: %v", err)
+			// TODO retry?
+		}
+		return trns, nil
+	} else { // remote
+		return srv.holder.FinishTransaction(id)
+	}
+
+}
+
+func (srv *Server) Transactions() (map[string]Transaction, error) {
+	node := srv.node()
+	if !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
+		return nil, ErrNodeNotCoordinator
+	}
+
+	return srv.holder.Transactions()
+}
+
+func (srv *Server) GetTransaction(id string, remote bool) (Transaction, error) {
+	node := srv.node()
+	if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
+		return Transaction{}, ErrNodeNotCoordinator
+	}
+
+	if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) {
+		return Transaction{}, errors.New("got a remote finish call to coordinator or single node cluster... shouldn't ever happen")
+	}
+
+	trns, err := srv.holder.GetTransaction(id)
+	if err != nil {
+		return Transaction{}, errors.Wrap(err, "getting transaction")
+	}
+
+	// The way a client would find out that the exclusive transaction
+	// it requested is active is by polling the GetTransaction
+	// endpoint. Therefore, returning an active, exclusive
+	// transaction, from here is what truly makes the transaction
+	// "live". Before doing so, we want to make sure all nodes
+	// agree. (in case other nodes have activity on this transaction
+	// we're not aware of)
+	if !remote && trns.Exclusive && trns.Active {
+		err := srv.SendSync(
+			&TransactionMessage{
+				Action:      TRANSACTION_VALIDATE,
+				Transaction: trns,
+			},
+		)
+		if err != nil {
+			return Transaction{}, errors.Wrap(err, "contacting remote hosts")
+		}
+		return trns, nil
+	} else { // remote
+		return trns, nil
+	}
+}
+
 // countOpenFiles on operating systems that support lsof.
 func countOpenFiles() (int, error) {
 	switch runtime.GOOS {

server/server.go

@@ -87,6 +87,7 @@ type Command struct {
 	listenURI    *pilosa.URI
 	tlsConfig    *tls.Config
 	closeTimeout time.Duration
+	noSleep      bool
 
 	serverOptions []pilosa.ServerOption
 }
@@ -114,6 +115,17 @@ func OptCommandConfig(config *Config) CommandOption {
 	}
 }
 
+// OptCommandNoSleep disables the 5 second sleep for non-coordinator
+// nodes on startup. See https://github.com/molecula/pilosa/issues/266
+// This option should only be used by tests, and expect it to be
+// deprecated.
+func OptCommandNoSleep() CommandOption {
+	return func(c *Command) error {
+		c.noSleep = true
+		return nil
+	}
+}
+
 // NewCommand returns a new instance of Main.
 func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command {
 	c := &Command{
@@ -149,7 +161,7 @@ func (m *Command) Start() (err error) {
 	if !m.API.Node().IsCoordinator {
 		// hack to give coordinator a head start
 		// TODO https://github.com/molecula/pilosa/issues/266
-		if len(m.Config.Gossip.Seeds) > 0 {
+		if len(m.Config.Gossip.Seeds) > 0 && !m.noSleep {
 			time.Sleep(5 * time.Second)
 		}
 	}

server/server_test.go

@@ -374,6 +374,115 @@ func TestConcurrentFieldCreation(t *testing.T) {
 	}
 }
 
+func TestTransactionsAPI(t *testing.T) {
+	cluster := test.MustRunCluster(t, 3)
+	defer cluster.Close()
+
+	api0 := cluster[0].API
+	api1 := cluster[1].API
+	//api2 := cluster[2].API
+
+	// can fetch empty transactions
+	if trnsMap, err := api0.Transactions(); err != nil {
+		t.Fatalf("getting transactions: %v", err)
+	} else if len(trnsMap) != 0 {
+		t.Fatalf("unexpectedly has transactions: %v", trnsMap)
+	}
+
+	// can't fetch transactions from non-coordinator
+	if _, err := api1.Transactions(); err != pilosa.ErrNodeNotCoordinator {
+		t.Errorf("api1 should return ErrNodeNotCoordinator when asked for transactions but got: %v", err)
+	}
+
+	// can start transaction
+	if trns, err := api0.StartTransaction("a", time.Minute, false, false); err != nil {
+		t.Errorf("couldn't start transaction: %v", err)
+	} else {
+		test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
+	}
+
+	// can retrieve transaction from other nodes with remote=true
+	if trns, err := api1.GetTransaction("a", true); err != nil {
+		t.Errorf("couldn't fetch transaction from other node with remote=true: %v", err)
+	} else {
+		test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
+	}
+
+	// can start transaction with blank id and get uuid back
+	id := ""
+	if trns, err := api0.StartTransaction(id, time.Minute, false, false); err != nil {
+		t.Errorf("couldn't start transaction: %v", err)
+	} else {
+		id = trns.ID
+		if len(id) != 36 { // UUID
+			t.Errorf("unexpected generated ID: %s", id)
+		}
+		test.CompareTransactions(t, pilosa.Transaction{ID: id, Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
+	}
+
+	// can't finish transaction on non-coordinator
+	if _, err := api1.FinishTransaction(id, false); err != pilosa.ErrNodeNotCoordinator {
+		t.Errorf("unexpected error is not ErrNodeNotCoordinator: %v", err)
+	}
+
+	// can finish transaction
+	if _, err := api0.FinishTransaction(id, false); err != nil {
+		t.Errorf("couldn't finish transaction: %v", err)
+	}
+
+	// can finish previous transaction
+	if _, err := api0.FinishTransaction("a", false); err != nil {
+		t.Errorf("couldn't finish transaction a: %v", err)
+	}
+
+	// can start exclusive transaction
+	if te, err := api0.StartTransaction("exc", time.Minute, true, false); err != nil {
+		t.Errorf("couldn't start exclusive transaction: %v", err)
+	} else if !te.Active {
+		t.Errorf("expected exclusive transaction to be active: %+v", te)
+	}
+
+	// can finish exclusive transaction
+	if _, err := api0.FinishTransaction("exc", false); err != nil {
+		t.Errorf("couldn't finish exclusive transaction: %v", err)
+	}
+
+	// can start transaction (with same name as previous finished transaction)
+	if trns, err := api0.StartTransaction("a", time.Minute, false, false); err != nil {
+		t.Errorf("couldn't start transaction: %v", err)
+	} else {
+		test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
+	}
+
+	// can start exclusive transaction and is not immediately active
+	if te, err := api0.StartTransaction("exc", time.Minute, true, false); err != nil {
+		t.Errorf("couldn't start exclusive transaction: %v", err)
+	} else if te.Active {
+		t.Errorf("expected exclusive transaction to be inactive: %+v", te)
+	}
+
+	// can finish non-exclusive transaction
+	if _, err := api0.FinishTransaction("a", false); err != nil {
+		t.Errorf("couldn't finish transaction a: %v", err)
+	}
+
+	// can poll exclusive transaction and is active
+	if trns, err := api0.GetTransaction("exc", false); err != nil {
+		t.Errorf("couldn't poll exclusive transaction: %v", err)
+	} else {
+		test.CompareTransactions(t, pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
+	}
+
+	// transaction is active on other nodes with remote=true
+	if trns, err := api1.GetTransaction("exc", true); err != nil {
+		t.Errorf("couldn't poll exclusive transaction: %v", err)
+	} else {
+		test.CompareTransactions(t, pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns)
+	}
+
+	// LATER, test deadline extension on non-coordinator blocks active, exclusive transaction being returned
+}
+
 func TestMain_RecalculateHashes(t *testing.T) {
 	const clusterSize = 5
 	cluster := test.MustRunCluster(t, clusterSize)

test/pilosa.go

@@ -65,6 +65,7 @@ func newCommand(opts ...server.CommandOption) *Command {
 	// does not fail on 32-bit systems.
 	opts = append([]server.CommandOption{
 		server.OptCommandCloseTimeout(time.Millisecond * 2),
+		server.OptCommandNoSleep(),
 	}, opts...)
 	m := &Command{commandOptions: opts}
 	m.Command = server.NewCommand(bytes.NewReader(nil), ioutil.Discard, ioutil.Discard, opts...)
@@ -409,6 +410,7 @@ func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "new cluster")
 	}
+
 	if err = cluster.Start(); err != nil {
 		return nil, errors.Wrap(err, "starting cluster")
 	}

test/transaction.go

@@ -0,0 +1,36 @@
+package test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pilosa/pilosa/v2"
+)
+
+// CompareTransactions errors describing how the
+// transactions differ (if at all). The deadlines need only be close
+// (within 3ms).
+func CompareTransactions(t *testing.T, trns1, trns2 pilosa.Transaction) {
+	t.Helper()
+	if trns1.ID != trns2.ID {
+		t.Errorf("IDs differ:\n%+v\n%+v", trns1, trns2)
+	}
+	if trns1.Active != trns2.Active {
+		t.Errorf("Actives differ:\n%+v\n%+v", trns1, trns2)
+	}
+	if trns1.Exclusive != trns2.Exclusive {
+		t.Errorf("Exclusives differ:\n%+v\n%+v", trns1, trns2)
+	}
+	if trns1.Timeout != trns2.Timeout {
+		t.Errorf("Timeouts differ:\n%+v\n%+v", trns1, trns2)
+	}
+
+	diff := trns1.Deadline.Sub(trns2.Deadline)
+
+	if diff > time.Millisecond*3 || diff < time.Millisecond*-3 {
+		t.Errorf("Deadlines differ by %v:\n%+v\n%+v", diff, trns1, trns2)
+	}
+	if trns1.Stats != trns2.Stats {
+		t.Errorf("Stats differ:\n%+v\n%+v", trns1, trns2)
+	}
+}

transaction.go

@@ -140,9 +140,8 @@ func (tm *TransactionManager) finish(id string) (Transaction, error) {
 	// After removing, check to see if we need to activate an exclusive transaction
 	trnsMap, err := tm.store.List()
 	if err != nil {
-		// returning an error here is weird because we've already
-		// removed the transaction
-		return trns, errors.Wrap(err, "listing transactions in Finish")
+		tm.log().Printf("error listing transactions in Finish: %v", err)
+		return trns, nil
 	}
 
 	if len(trnsMap) == 1 {
@@ -154,7 +153,8 @@ func (tm *TransactionManager) finish(id string) (Transaction, error) {
 				etrans.Active = true
 				etrans.Deadline = time.Now().Add(etrans.Timeout)
 				if err := tm.store.Put(etrans); err != nil {
-					return trns, errors.Wrap(err, "activating exclusive transaction after finishing last transaction")
+					tm.log().Printf("activating exclusive transaction after finishing last transaction: %v", err)
+					return trns, nil
 				}
 			}
 		}
@@ -357,3 +357,20 @@ const ErrTransactionNotFound = Error("transaction not found")
 const ErrTransactionExclusive = Error("there is already an exclusive transaction")
 const ErrTransactionExists = Error("transaction with the given id already exists")
 const ErrTransactionInactive = Error("cannot finish an inactive transaction")
+
+func CompareTransactions(t1, t2 Transaction) error {
+	if t1.ID != t2.ID {
+		return errors.Errorf("transaction IDs not equal: %+v %+v", t1, t2)
+	}
+	if t1.Active != t2.Active {
+		return errors.Errorf("transaction Actives not equal: %+v %+v", t1, t2)
+	}
+	if t1.Exclusive != t2.Exclusive {
+		return errors.Errorf("transaction Exclusives not equal: %+v %+v", t1, t2)
+	}
+	if t1.Timeout != t2.Timeout {
+		return errors.Errorf("transaction Timeouts not equal: %+v %+v", t1, t2)
+	}
+	// don't care about Deadline or Stats
+	return nil
+}

transaction.md

@@ -4,7 +4,7 @@ This is not full-featured transaction support with commit and rollback
 for now; this is a placeholder intended to allow us to solve shorter-term
 problems.
 
-The primaryw purpose of this is to allow an exclusive transaction to
+The primary purpose of this is to allow an exclusive transaction to
 block new ingest activity from starting, while permitting existing ingest
 operations to complete, even if a single ingest requires multiple operations.
 This allows users with cooperating ingest operations to ensure a stable state
@@ -109,3 +109,64 @@ If multiple exclusive transactions are requested, they become active
 sequentially in the order the requests came in, and the snapshot queue and
 other transactions are not permitted to resume until the exclusive transactions
 all complete.
+
+
+
+### Implementation Notes
+
+All requests go through coordinator.
+
+When creating a new transaction, we'll create it on every node in the
+cluster and persist it to disk.
+
+Only the coordinator will accept requests to start a transaction.
+
+Timeouts only expire when there has been *no activity* on a transaction for the timeout duration. 
+Any activity on the transaction may extend the deadline (unimplemented).
+
+When finishing a transaction, we'll finish it on the coordinator and
+then broadcast the finish to the cluster before returning to the
+client.
+
+When getting an exclusive transaction, if the transaction is active,
+we'll make sure that all nodes agree before returning it.
+
+
+Coordinator forwards all requests to every other node so they can stay
+in sync. If the coordinator doesn't hear back from a node, the request
+fails. The coordinator only reaches out to active nodes, so if the
+cluster is in DEGRADED, things can still continue.
+
+If an node is down and comes back up it needs to synchronize its state
+with the coordinator (unimplemented).
+
+There is a separate TransactionManager and TransactionStore
+
+The store is just responsible for persisting info about
+transactions. The manager handles all the logic (at the node level).
+Logic related to cluster and remote vs local node is handled by the
+Server. The Holder contains the TransactionManager, and the Server
+contains the logic for how to handle external vs intra cluster
+requests (remote==true).
+
+There is intra-cluster messaging for transactions which is handled
+with the new TransactionMessage and goes through the usual
+SendMessage/Broadcaster stuff.
+
+There is also external API which is handled by the HTTP handler and
+goes through API (and is passed directly to Server). (unimplemented)
+
+
+#### TODO
+
+- [x] implement api layer and cluster logic, startup, etc. 
+- [ ] implement HTTP layer including header/transaction ID
+- [ ] implement and use persistent transaction store rather than inmem.
+- [ ] update go-pilosa/gpexp to actually USE transactions
+  - [ ] update IDK to use updated go-pilosa
+
+#### Testing TransactionManager
+- there should never be more than one Exclusive transaction
+- if the Exclusive transaction is active, there should be no other transactions
+
+

transaction_test.go

@@ -20,34 +20,34 @@ func TestTransactionManager(t *testing.T) {
 
 	// can add a non-exclusive transaction
 	trns1 := mustStart(t, tm, "a", time.Microsecond, false)
-	compareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns1)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns1)
 
 	// can have two non exclusive transactions
 	trns2 := mustStart(t, tm, "b", time.Microsecond, false)
-	compareTransactions(t, pilosa.Transaction{ID: "b", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns2)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "b", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns2)
 
 	// trying to start a transaction with same name errors and returns previous transaction
 	t3, err := tm.Start("a", time.Second, true)
 	if err != pilosa.ErrTransactionExists {
 		t.Errorf("expected transaction exists, but got: '%v'", err)
 	}
-	compareTransactions(t, trns1, t3)
+	test.CompareTransactions(t, trns1, t3)
 
 	// can get an existing transaction
 	trns2_2 := mustGet(t, tm, "b")
-	compareTransactions(t, trns2, trns2_2)
+	test.CompareTransactions(t, trns2, trns2_2)
 
 	// can list all transactions
 	trnsMap := mustList(t, tm)
 	if len(trnsMap) != 2 {
 		t.Errorf("unexpected number of transactions in map: %d", len(trnsMap))
 	}
-	compareTransactions(t, trnsMap["a"], trns1)
-	compareTransactions(t, trnsMap["b"], trns2)
+	test.CompareTransactions(t, trnsMap["a"], trns1)
+	test.CompareTransactions(t, trnsMap["b"], trns2)
 
 	// can submit an exclusive transaction
 	trnsE := mustStart(t, tm, "ce", time.Millisecond*5, true)
-	compareTransactions(t, pilosa.Transaction{ID: "ce", Active: false, Exclusive: true, Timeout: time.Millisecond * 5, Deadline: time.Now().Add(time.Millisecond * 5)}, trnsE)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "ce", Active: false, Exclusive: true, Timeout: time.Millisecond * 5, Deadline: time.Now().Add(time.Millisecond * 5)}, trnsE)
 
 	// can't start new transactions while an exclusive transaction is pending
 	if _, err := tm.Start("d", time.Millisecond, false); err != pilosa.ErrTransactionExclusive {
@@ -101,7 +101,7 @@ func TestTransactionManager(t *testing.T) {
 
 	// can start a new exclusive transaction and it's immediately active
 	trnsHE := mustStart(t, tm, "he", time.Hour, true)
-	compareTransactions(t, pilosa.Transaction{ID: "he", Active: true, Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsHE)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "he", Active: true, Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsHE)
 
 	// can't start new transactions while an exclusive transaction is active
 	if _, err := tm.Start("i", time.Millisecond, false); err != pilosa.ErrTransactionExclusive {
@@ -110,40 +110,40 @@ func TestTransactionManager(t *testing.T) {
 
 	// can finish an active exclusive transaction
 	trnsHE_finish := mustFinish(t, tm, "he")
-	compareTransactions(t, trnsHE, trnsHE_finish)
+	test.CompareTransactions(t, trnsHE, trnsHE_finish)
 
 	// can start normal transaction after finishing exclusive transaction
 	trnsJ := mustStart(t, tm, "j", time.Hour, false)
-	compareTransactions(t, pilosa.Transaction{ID: "j", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsJ)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "j", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsJ)
 
 	// can finish normal transaction
 	trnsJ_finish := mustFinish(t, tm, "j")
-	compareTransactions(t, trnsJ, trnsJ_finish)
+	test.CompareTransactions(t, trnsJ, trnsJ_finish)
 
 	// can start normal transaction after finishing normal transaction
 	trnsK := mustStart(t, tm, "k", time.Hour, false)
-	compareTransactions(t, pilosa.Transaction{ID: "k", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsK)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "k", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsK)
 
 	// can start new exclusive transaction, but not immediately active
 	trnsLE := mustStart(t, tm, "le", time.Hour, true)
-	compareTransactions(t, pilosa.Transaction{ID: "le", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsLE)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "le", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsLE)
 
 	// finishing k should activate le
 	trnsK_finish := mustFinish(t, tm, "k")
-	compareTransactions(t, trnsK, trnsK_finish)
+	test.CompareTransactions(t, trnsK, trnsK_finish)
 	trnsLE_active := mustGet(t, tm, "le")
 	trnsLE.Active = true
-	compareTransactions(t, trnsLE, trnsLE_active)
+	test.CompareTransactions(t, trnsLE, trnsLE_active)
 
 	mustFinish(t, tm, "le")
 
 	// can start normal transaction to test deadline reset
 	trnsM := mustStart(t, tm, "m", time.Millisecond*4, false)
-	compareTransactions(t, pilosa.Transaction{ID: "m", Active: true, Timeout: time.Millisecond * 4, Deadline: time.Now().Add(time.Millisecond * 4)}, trnsM)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "m", Active: true, Timeout: time.Millisecond * 4, Deadline: time.Now().Add(time.Millisecond * 4)}, trnsM)
 
 	// start new exclusive transaction to trigger deadline check
 	trnsNE := mustStart(t, tm, "ne", time.Hour, true)
-	compareTransactions(t, pilosa.Transaction{ID: "ne", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsNE)
+	test.CompareTransactions(t, pilosa.Transaction{ID: "ne", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsNE)
 
 	// sleep for most of the deadline
 	time.Sleep(time.Millisecond * 3)
@@ -154,14 +154,14 @@ func TestTransactionManager(t *testing.T) {
 		t.Errorf("resetting deadline: %v", err)
 	}
 	trnsM.Deadline = time.Now().Add(time.Millisecond * 4)
-	compareTransactions(t, trnsM, trnsM_reset)
+	test.CompareTransactions(t, trnsM, trnsM_reset)
 
 	// sleep until past the original deadline
 	time.Sleep(time.Millisecond * 2)
 
 	// verify that trnsM still exists
 	trnsM_again := mustGet(t, tm, "m")
-	compareTransactions(t, trnsM, trnsM_again)
+	test.CompareTransactions(t, trnsM, trnsM_again)
 
 }
 
@@ -201,34 +201,6 @@ func mustList(t *testing.T, tm *pilosa.TransactionManager) map[string]pilosa.Tra
 	return trnsMap
 }
 
-// compareTransactions errors describing how the
-// transactions differ (if at all). The deadlines need only be close
-// (within 3ms).
-func compareTransactions(t *testing.T, trns1, trns2 pilosa.Transaction) {
-	t.Helper()
-	if trns1.ID != trns2.ID {
-		t.Errorf("IDs differ:\n%+v\n%+v", trns1, trns2)
-	}
-	if trns1.Active != trns2.Active {
-		t.Errorf("Actives differ:\n%+v\n%+v", trns1, trns2)
-	}
-	if trns1.Exclusive != trns2.Exclusive {
-		t.Errorf("Exclusives differ:\n%+v\n%+v", trns1, trns2)
-	}
-	if trns1.Timeout != trns2.Timeout {
-		t.Errorf("Timeouts differ:\n%+v\n%+v", trns1, trns2)
-	}
-
-	diff := trns1.Deadline.Sub(trns2.Deadline)
-
-	if diff > time.Millisecond*3 || diff < time.Millisecond*-3 {
-		t.Errorf("Deadlines differ by %v:\n%+v\n%+v", diff, trns1, trns2)
-	}
-	if trns1.Stats != trns2.Stats {
-		t.Errorf("Stats differ:\n%+v\n%+v", trns1, trns2)
-	}
-}
-
 func TestInMemTransactionStore(t *testing.T) {
 	ims := pilosa.NewInMemTransactionStore()