add HTTP handlers and client for transactions

client.go

@@ -17,6 +17,7 @@ package pilosa
 import (
 	"context"
 	"io"
+	"time"
 )
 
 // Bit represents the intersection of a row and a column. It can be specified by
@@ -73,6 +74,11 @@ type InternalClient interface {
 	RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error)
 	ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
 	ImportColumnAttrs(ctx context.Context, uri *URI, index string, req *ImportColumnAttrsRequest) error
+
+	StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (Transaction, error)
+	FinishTransaction(ctx context.Context, id string) (Transaction, error)
+	Transactions(ctx context.Context) (map[string]Transaction, error)
+	GetTransaction(ctx context.Context, id string) (Transaction, error)
 }
 
 //===============
@@ -204,3 +210,16 @@ func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, fiel
 func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) {
 	return nil, nil
 }
+
+func (n nopInternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (Transaction, error) {
+	return Transaction{}, nil
+}
+func (n nopInternalClient) FinishTransaction(ctx context.Context, id string) (Transaction, error) {
+	return Transaction{}, nil
+}
+func (n nopInternalClient) Transactions(ctx context.Context) (map[string]Transaction, error) {
+	return nil, nil
+}
+func (n nopInternalClient) GetTransaction(ctx context.Context, id string) (Transaction, error) {
+	return Transaction{}, nil
+}

http/client.go

@@ -26,6 +26,7 @@ import (
 	"net/url"
 	"sort"
 	"strconv"
+	"time"
 
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/encoding/proto"
@@ -392,6 +393,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits
 
 	// Get the coordinator node; all bits are sent to the
 	// primary translate store (i.e. coordinator).
+	// TODO... is that right^^?
 	nodes, err := c.Nodes(ctx)
 	if err != nil {
 		return fmt.Errorf("getting nodes: %s", err)
@@ -1227,11 +1229,165 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI,
 	return tkresp.Keys, nil
 }
 
+func (c *InternalClient) Transactions(ctx context.Context) (map[string]pilosa.Transaction, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Transactions")
+	defer span.Finish()
+
+	trnsMap := make(map[string]pilosa.Transaction)
+
+	u := uriPathToURL(c.defaultURI, "/transactions")
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return trnsMap, errors.Wrap(err, "creating transactions request")
+	}
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
+
+	resp, err := c.executeRequest(req.WithContext(ctx))
+	if err != nil {
+		return trnsMap, errors.Wrap(err, "executing request")
+	}
+	defer resp.Body.Close()
+	tmpTrnsMap := make(map[string]*pilosa.Transaction)
+	err = json.NewDecoder(resp.Body).Decode(&tmpTrnsMap)
+
+	for id, trnsp := range tmpTrnsMap {
+		trnsMap[id] = *trnsp
+	}
+
+	return trnsMap, errors.Wrap(err, "json decoding")
+}
+
+func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (pilosa.Transaction, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.StartTransaction")
+	defer span.Finish()
+	tr := &TransactionResponse{Transaction: &pilosa.Transaction{}}
+	buf, err := json.Marshal(&pilosa.Transaction{
+		ID:        id,
+		Timeout:   timeout,
+		Exclusive: exclusive,
+	})
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "marshalling payload")
+	}
+	// We're using the defaultURI here because this is only used by
+	// tests, and we want to test requests against all hosts. A robust
+	// client implementation would ensure that these requests go to
+	// the coordinator.
+	u := uriPathToURL(c.defaultURI, "/transaction/"+id)
+	req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "creating post transaction request")
+	}
+	req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
+
+	resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "executing request")
+	}
+	defer resp.Body.Close()
+	err = json.NewDecoder(resp.Body).Decode(&tr)
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "decoding response")
+	}
+	if resp.StatusCode == 409 {
+		err = pilosa.ErrTransactionExclusive
+	} else if tr.Error != "" {
+		err = errors.New(tr.Error)
+	}
+	return *tr.Transaction, err
+}
+
+func (c *InternalClient) FinishTransaction(ctx context.Context, id string) (pilosa.Transaction, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FinishTransaction")
+	defer span.Finish()
+
+	u := uriPathToURL(c.defaultURI, "/transaction/"+id+"/finish")
+	req, err := http.NewRequest("POST", u.String(), nil)
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "creating finish transaction request")
+	}
+
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
+
+	resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "executing request")
+	}
+	defer resp.Body.Close()
+	tr := &TransactionResponse{Transaction: &pilosa.Transaction{}}
+	err = json.NewDecoder(resp.Body).Decode(&tr)
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "decoding response")
+	}
+
+	if tr.Error != "" {
+		err = errors.New(tr.Error)
+	}
+	return *tr.Transaction, err
+}
+
+func (c *InternalClient) GetTransaction(ctx context.Context, id string) (pilosa.Transaction, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.GetTransaction")
+	defer span.Finish()
+
+	// We're using the defaultURI here because this is only used by
+	// tests, and we want to test requests against all hosts. A robust
+	// client implementation would ensure that these requests go to
+	// the coordinator.
+	u := uriPathToURL(c.defaultURI, "/transaction/"+id)
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "creating get transaction request")
+	}
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
+
+	resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "executing request")
+	}
+	defer resp.Body.Close()
+	tr := &TransactionResponse{Transaction: &pilosa.Transaction{}}
+	err = json.NewDecoder(resp.Body).Decode(&tr)
+	if err != nil {
+		return pilosa.Transaction{}, errors.Wrap(err, "decoding response")
+	}
+
+	if tr.Error != "" {
+		err = errors.New(tr.Error)
+	}
+	return *tr.Transaction, err
+}
+
+type executeOpts struct {
+	// giveRawResponse instructs executeRequest not to process the
+	// respStatusCode and try to extract errors or whatever.
+	giveRawResponse bool
+}
+
+type executeRequestOption func(*executeOpts)
+
+func giveRawResponse(b bool) executeRequestOption {
+	return func(eo *executeOpts) {
+		eo.giveRawResponse = b
+	}
+}
+
 // executeRequest executes the given request and checks the Response. For
 // responses with non-2XX status, the body is read and closed, and an error is
 // returned. If the error is nil, the caller must ensure that the response body
 // is closed.
-func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, error) {
+func (c *InternalClient) executeRequest(req *http.Request, opts ...executeRequestOption) (*http.Response, error) {
+	eo := &executeOpts{}
+	for _, opt := range opts {
+		opt(eo)
+	}
+
 	tracing.GlobalTracer.InjectHTTPHeaders(req)
 	req.Close = false
 	resp, err := c.httpClient.Do(req)
@@ -1241,6 +1397,9 @@ func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, erro
 		}
 		return nil, errors.Wrap(err, "getting response")
 	}
+	if eo.giveRawResponse {
+		return resp, nil
+	}
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 		defer resp.Body.Close()
 		buf, err := ioutil.ReadAll(resp.Body)

http/client_test.go

@@ -23,6 +23,7 @@ import (
 	gohttp "net/http"
 	"reflect"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -32,6 +33,7 @@ import (
 	"github.com/pilosa/pilosa/v2/pql"
 	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
+	"github.com/pkg/errors"
 )
 
 // Test distributed TopN Row count across 3 nodes.
@@ -1242,6 +1244,150 @@ func TestClient_CreateDecimalField(t *testing.T) {
 	}
 }
 
+func TestClientTransactions(t *testing.T) {
+	c := test.MustRunCluster(t, 3)
+	defer c.Close()
+
+	client0 := MustNewClient(c[0].URL(), http.GetHTTPClient(nil))
+	client1 := MustNewClient(c[1].URL(), http.GetHTTPClient(nil))
+
+	// can create, list, get, and finish a transaction
+	var expDeadline time.Time
+	if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); err != nil {
+		t.Fatalf("error starting transaction: %v", err)
+	} else {
+		expDeadline = time.Now().Add(time.Minute)
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline},
+			trns)
+	}
+
+	if trnsMap, err := client0.Transactions(context.Background()); err != nil {
+		t.Errorf("listing transactions: %v", err)
+	} else {
+		if len(trnsMap) != 1 {
+			t.Errorf("unexpected trnsMap: %+v", trnsMap)
+		}
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline},
+			trnsMap["blah"])
+	}
+
+	if trns, err := client0.GetTransaction(context.Background(), "blah"); err != nil {
+		t.Fatalf("error getting transaction: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline},
+			trns)
+	}
+
+	if trns, err := client0.FinishTransaction(context.Background(), "blah"); err != nil {
+		t.Fatalf("error finishing transaction: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// can create exclusive transaction
+	if trns, err := client0.StartTransaction(context.Background(), "blahe", time.Minute, true); err != nil {
+		t.Fatalf("error starting transaction: %v", err)
+	} else {
+		expDeadline = time.Now().Add(time.Minute)
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: true, Exclusive: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// cannot start new transaction - correct error and exclusive transaction are returned
+	if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); errors.Cause(err) != pilosa.ErrTransactionExclusive {
+		t.Fatalf("shouldn't be able to start transaction while an exclusive is running, but got: %+v, %v", trns, err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: true, Exclusive: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// finish exclusive transaction
+	if trns, err := client0.FinishTransaction(context.Background(), "blahe"); err != nil {
+		t.Fatalf("error finishing transaction: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: true, Exclusive: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// start new transaction
+	if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); err != nil {
+		t.Fatalf("error starting transaction: %v", err)
+	} else {
+		expDeadline = time.Now().Add(time.Minute)
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// try to start same transaction
+	if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); err == nil ||
+		!strings.Contains(err.Error(), pilosa.ErrTransactionExists.Error()) {
+		t.Fatalf("expected ErrTransactionExists, but got: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// start an exclusive transaction which can't go active
+	if trns, err := client0.StartTransaction(context.Background(), "blahe", time.Minute, true); err != nil {
+		t.Fatalf("error starting transaction: %v", err)
+	} else {
+		expDeadline = time.Now().Add(time.Minute)
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: false, Exclusive: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// finish exclusive transaction that never went active
+	if trns, err := client0.FinishTransaction(context.Background(), "blahe"); err != nil {
+		t.Fatalf("error finishing transaction: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: false, Exclusive: true, Deadline: expDeadline},
+			trns)
+	}
+
+	// finish non-existent transaction
+	if trns, err := client0.FinishTransaction(context.Background(), "zzz"); err == nil ||
+		!strings.Contains(err.Error(), pilosa.ErrTransactionNotFound.Error()) {
+		t.Fatalf("unexpected error finishing nonexistent transaction: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{},
+			trns)
+	}
+
+	// get non-existent transaction
+	if trns, err := client0.GetTransaction(context.Background(), "xxx"); err == nil ||
+		!strings.Contains(err.Error(), pilosa.ErrTransactionNotFound.Error()) {
+		t.Fatalf("unexpected error getting nonexistent transaction: %v", err)
+	} else {
+		test.CompareTransactions(t,
+			pilosa.Transaction{},
+			trns)
+	}
+
+	// non-coordinator
+	if trns, err := client1.StartTransaction(context.Background(), "blah", time.Minute, false); err == nil ||
+		!strings.Contains(err.Error(), pilosa.ErrNodeNotCoordinator.Error()) {
+		t.Fatalf("unexpected error starting on non-coordinator: %v", err)
+	} else {
+		expDeadline = time.Now().Add(time.Minute)
+		test.CompareTransactions(t,
+			pilosa.Transaction{},
+			trns)
+	}
+}
+
 // Client represents a test wrapper for pilosa.Client.
 type Client struct {
 	*http.InternalClient

http/handler.go

@@ -204,6 +204,10 @@ func (h *Handler) populateValidators() {
 	h.validators["PostFieldAttrDiff"] = queryValidationSpecRequired()
 	h.validators["GetNodes"] = queryValidationSpecRequired()
 	h.validators["GetShardMax"] = queryValidationSpecRequired()
+	h.validators["GetTransactions"] = queryValidationSpecRequired()
+	h.validators["GetTransaction"] = queryValidationSpecRequired()
+	h.validators["PostTransaction"] = queryValidationSpecRequired()
+	h.validators["PostFinishTransaction"] = queryValidationSpecRequired()
 }
 
 type contextKeyQuery int
@@ -337,6 +341,10 @@ func newRouter(handler *Handler) *mux.Router {
 	router.HandleFunc("/schema", handler.handlePostSchema).Methods("POST").Name("PostSchema")
 	router.HandleFunc("/status", handler.handleGetStatus).Methods("GET").Name("GetStatus")
 	router.HandleFunc("/version", handler.handleGetVersion).Methods("GET").Name("GetVersion")
+	router.HandleFunc("/transactions", handler.handleGetTransactions).Methods("GET").Name("GetTransactions")
+	router.HandleFunc("/transaction/{id}", handler.handleGetTransaction).Methods("GET").Name("GetTransaction")
+	router.HandleFunc("/transaction/{id}", handler.handlePostTransaction).Methods("POST").Name("PostTransaction")
+	router.HandleFunc("/transaction/{id}/finish", handler.handlePostFinishTransaction).Methods("POST").Name("PostFinishTransaction")
 
 	// /internal endpoints are for internal use only; they may change at any time.
 	// DO NOT rely on these for external applications!
@@ -1031,6 +1039,107 @@ func (h *Handler) handleDeleteField(w http.ResponseWriter, r *http.Request) {
 	resp.write(w, err)
 }
 
+func (h *Handler) handleGetTransactions(w http.ResponseWriter, r *http.Request) {
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
+		return
+	}
+	trnsMap, err := h.api.Transactions()
+	if err != nil {
+		switch errors.Cause(err) {
+		case pilosa.ErrNodeNotCoordinator:
+			http.Error(w, err.Error(), http.StatusBadRequest)
+		default:
+			http.Error(w, "problem getting transactions: "+err.Error(), http.StatusInternalServerError)
+		}
+		return
+	}
+
+	// JSON marshalling bullshit. Maybe we should just use
+	// *Transaction everywhere.
+	tmapP := make(map[string]*pilosa.Transaction)
+	for id, trns := range trnsMap {
+		trns := trns
+		tmapP[id] = &trns
+	}
+
+	if err := json.NewEncoder(w).Encode(tmapP); err != nil {
+		h.logger.Printf("encoding GetTransactions response: %s", err)
+	}
+}
+
+type TransactionResponse struct {
+	Transaction *pilosa.Transaction `json:"transaction,omitempty"`
+	Error       string              `json:"error,omitempty"`
+}
+
+func (h *Handler) doTransactionResponse(w http.ResponseWriter, err error, trns pilosa.Transaction) {
+	if err != nil {
+		switch errors.Cause(err) {
+		case pilosa.ErrNodeNotCoordinator, pilosa.ErrTransactionExists:
+			w.WriteHeader(http.StatusBadRequest)
+		case pilosa.ErrTransactionExclusive:
+			w.WriteHeader(http.StatusConflict)
+		case pilosa.ErrTransactionNotFound:
+			w.WriteHeader(http.StatusNotFound)
+		default:
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+	}
+
+	var errString string
+	if err != nil {
+		errString = err.Error()
+	}
+	err = json.NewEncoder(w).Encode(
+		TransactionResponse{Error: errString, Transaction: &trns})
+	if err != nil {
+		h.logger.Printf("encoding transaction response: %v", err)
+	}
+
+}
+
+func (h *Handler) handleGetTransaction(w http.ResponseWriter, r *http.Request) {
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
+		return
+	}
+	id := mux.Vars(r)["id"]
+	trns, err := h.api.GetTransaction(id, false)
+	h.doTransactionResponse(w, err, trns)
+}
+
+func (h *Handler) handlePostTransaction(w http.ResponseWriter, r *http.Request) {
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
+		return
+	}
+	reqTrns := &pilosa.Transaction{}
+	if err := json.NewDecoder(r.Body).Decode(reqTrns); err != nil || reqTrns.Timeout == 0 {
+		if err == nil {
+			http.Error(w, "timeout is required and cannot be 0", http.StatusBadRequest)
+		} else {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+		}
+		return
+	}
+
+	id := mux.Vars(r)["id"]
+	trns, err := h.api.StartTransaction(id, reqTrns.Timeout, reqTrns.Exclusive, false)
+
+	h.doTransactionResponse(w, err, trns)
+}
+
+func (h *Handler) handlePostFinishTransaction(w http.ResponseWriter, r *http.Request) {
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
+		return
+	}
+	id := mux.Vars(r)["id"]
+	trns, err := h.api.FinishTransaction(id, false)
+	h.doTransactionResponse(w, err, trns)
+}
+
 // handleDeleteRemoteAvailableShard handles DELETE /field/{field}/available-shards/{shardID} request.
 func (h *Handler) handleDeleteRemoteAvailableShard(w http.ResponseWriter, r *http.Request) {
 	if !validHeaderAcceptJSON(r.Header) {

transaction.go

@@ -1,6 +1,7 @@
 package pilosa
 
 import (
+	"encoding/json"
 	"sync"
 	"time"
 
@@ -12,25 +13,25 @@ import (
 // needs to be tracked and spans multiple API calls.
 type Transaction struct {
 	// ID is an arbitrary string identifier. All transactions must have a unique ID.
-	ID string
+	ID string `json:"id"`
 
 	// Active notes whether an Exclusive transaction is active, or
 	// still pending (if other active transactions exist). All
 	// non-exclusive transactions are always active.
-	Active bool
+	Active bool `json:"active"`
 
 	// Exclusive is set on transactions which can only become active when no other transactions exist.
-	Exclusive bool
+	Exclusive bool `json:"exclusive"`
 
 	// Timeout is the minimum idle time for which this transaction should continue to exist.
-	Timeout time.Duration
+	Timeout time.Duration `json:"timeout"`
 
 	// Deadline is calculated from Timeout, and should be reset each
 	// time there is activity on the transaction.
-	Deadline time.Time
+	Deadline time.Time `json:"deadline"`
 
 	// Stats track statistics for the transaction. Not yet used.
-	Stats TransactionStats
+	Stats TransactionStats `json:"stats"`
 }
 
 type TransactionStats struct{}
@@ -86,7 +87,7 @@ func (tm *TransactionManager) Start(id string, timeout time.Duration, exclusive
 			// if someone wants a transaction, and we're not able to
 			// give it to them, we want to be checking deadlines.
 			tm.startDeadlineChecker()
-			return Transaction{}, ErrTransactionExclusive
+			return trns, ErrTransactionExclusive
 		}
 	}
 	if trns, ok := trnsMap[id]; ok {
@@ -354,9 +355,8 @@ type Error string
 func (e Error) Error() string { return string(e) }
 
 const ErrTransactionNotFound = Error("transaction not found")
-const ErrTransactionExclusive = Error("there is already an exclusive transaction")
+const ErrTransactionExclusive = Error("there is an exclusive transaction, try later")
 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 {
@@ -374,3 +374,58 @@ func CompareTransactions(t1, t2 Transaction) error {
 	// don't care about Deadline or Stats
 	return nil
 }
+
+func (trns *Transaction) UnmarshalJSON(b []byte) error {
+	tmp := &struct {
+		ID        string      `json:"id"`
+		Active    bool        `json:"active"`
+		Exclusive bool        `json:"exclusive"`
+		Timeout   interface{} `json:"timeout"`
+		Deadline  string      `json:"deadline"`
+	}{}
+	err := json.Unmarshal(b, tmp)
+	if err != nil {
+		return err
+	}
+	trns.ID = tmp.ID
+	trns.Active = tmp.Active
+	trns.Exclusive = tmp.Exclusive
+	switch tm := tmp.Timeout.(type) {
+	case string:
+		dur, err := time.ParseDuration(tm)
+		if err != nil {
+			return errors.Wrapf(err, "timeout as string must be a valid duration got: '%s'", tm)
+		}
+		trns.Timeout = dur
+	case float64:
+		// interpret as number of seconds
+		seconds := int64(tm)
+		nsec := (tm - float64(seconds)) * 1e9
+		trns.Timeout = time.Duration(seconds*1e9 + int64(nsec))
+	case nil:
+		break
+	default:
+		return errors.New("timeout must be float64 or string")
+	}
+
+	if tmp.Deadline != "" {
+		trns.Deadline, err = time.Parse(time.RFC3339Nano, tmp.Deadline)
+	}
+	return errors.Wrap(err, "parsing deadline")
+}
+
+func (trns *Transaction) MarshalJSON() ([]byte, error) {
+	return json.Marshal(&struct {
+		ID        string `json:"id"`
+		Active    bool   `json:"active"`
+		Exclusive bool   `json:"exclusive"`
+		Timeout   string `json:"timeout"`
+		Deadline  string `json:"deadline"`
+	}{
+		ID:        trns.ID,
+		Active:    trns.Active,
+		Exclusive: trns.Exclusive,
+		Timeout:   trns.Timeout.String(),
+		Deadline:  trns.Deadline.Format(time.RFC3339Nano),
+	})
+}

transaction.md

@@ -164,6 +164,8 @@ goes through API (and is passed directly to Server). (unimplemented)
 - [ ] implement and use persistent transaction store rather than inmem.
 - [ ] update go-pilosa/gpexp to actually USE transactions
   - [ ] update IDK to use updated go-pilosa
+  
+- ID validation. No slashes, no non-URL safe chars
 
 #### Testing TransactionManager
 - there should never be more than one Exclusive transaction

transaction_test.go

@@ -1,6 +1,7 @@
 package pilosa_test
 
 import (
+	"encoding/json"
 	"testing"
 	"time"
 
@@ -234,3 +235,76 @@ func TestInMemTransactionStore(t *testing.T) {
 	}
 
 }
+
+func TestMarshalUnmarshalTransaction(t *testing.T) {
+	tests := []struct {
+		name        string
+		transaction pilosa.Transaction
+	}{
+		{
+			name: "empty",
+		},
+		{
+			name: "basic",
+			transaction: pilosa.Transaction{
+				ID:        "blah",
+				Active:    true,
+				Exclusive: true,
+				Timeout:   time.Minute,
+				Deadline:  time.Now(),
+			},
+		},
+	}
+
+	for _, tst := range tests {
+		t.Run(tst.name, func(t *testing.T) {
+			bytes, err := json.Marshal(&tst.transaction)
+			if err != nil {
+				t.Errorf("marshalling: %v", err)
+			}
+
+			nt := &pilosa.Transaction{}
+			err = json.Unmarshal(bytes, nt)
+			if err != nil {
+				t.Fatalf("unmarshalling: %v", err)
+			}
+
+			test.CompareTransactions(t, tst.transaction, *nt)
+		})
+	}
+}
+
+func TestUnmarshalTransaction(t *testing.T) {
+	tests := []struct {
+		name            string
+		transactionJSON string
+		exp             pilosa.Transaction
+	}{
+		{
+			name:            "empty",
+			transactionJSON: `{}`,
+		},
+		{
+			name:            "basicPost",
+			transactionJSON: `{"id": "blah", "exclusive": false, "timeout": "1m"}`,
+			exp:             pilosa.Transaction{ID: "blah", Timeout: time.Minute},
+		},
+		{
+			name:            "basicPostFloatTimeout",
+			transactionJSON: `{"id": "blah", "exclusive": false, "timeout": 10.5}`,
+			exp:             pilosa.Transaction{ID: "blah", Timeout: time.Second*10 + time.Second/2},
+		},
+	}
+
+	for _, tst := range tests {
+		t.Run(tst.name, func(t *testing.T) {
+			nt := &pilosa.Transaction{}
+			err := json.Unmarshal([]byte(tst.transactionJSON), nt)
+			if err != nil {
+				t.Fatalf("unmarshalling: %v", err)
+			}
+
+			test.CompareTransactions(t, tst.exp, *nt)
+		})
+	}
+}