mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-10-10 21:07:53 +00:00
propagate context throughout transaction stuff
This commit is contained in:
Matt Jaffee
parent
41975de6b8
commit
c36952a0f1
8 changed files with 76 additions and 68 deletions
16
api.go
16
api.go
|
|
@ -1585,20 +1585,20 @@ 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) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool, remote bool) (Transaction, error) {
|
||||
return api.server.StartTransaction(ctx, id, timeout, exclusive, remote)
|
||||
}
|
||||
|
||||
func (api *API) FinishTransaction(id string, remote bool) (Transaction, error) {
|
||||
return api.server.FinishTransaction(id, remote)
|
||||
func (api *API) FinishTransaction(ctx context.Context, id string, remote bool) (Transaction, error) {
|
||||
return api.server.FinishTransaction(ctx, id, remote)
|
||||
}
|
||||
|
||||
func (api *API) Transactions() (map[string]Transaction, error) {
|
||||
return api.server.Transactions()
|
||||
func (api *API) Transactions(ctx context.Context) (map[string]Transaction, error) {
|
||||
return api.server.Transactions(ctx)
|
||||
}
|
||||
|
||||
func (api *API) GetTransaction(id string, remote bool) (Transaction, error) {
|
||||
return api.server.GetTransaction(id, remote)
|
||||
func (api *API) GetTransaction(ctx context.Context, id string, remote bool) (Transaction, error) {
|
||||
return api.server.GetTransaction(ctx, id, remote)
|
||||
}
|
||||
|
||||
type serverInfo struct {
|
||||
|
|
|
|||
16
holder.go
16
holder.go
|
|
@ -104,20 +104,20 @@ 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) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (Transaction, error) {
|
||||
return h.transactionManager.Start(ctx, id, timeout, exclusive)
|
||||
}
|
||||
|
||||
func (h *Holder) FinishTransaction(id string) (Transaction, error) {
|
||||
return h.transactionManager.Finish(id)
|
||||
func (h *Holder) FinishTransaction(ctx context.Context, id string) (Transaction, error) {
|
||||
return h.transactionManager.Finish(ctx, id)
|
||||
}
|
||||
|
||||
func (h *Holder) Transactions() (map[string]Transaction, error) {
|
||||
return h.transactionManager.List()
|
||||
func (h *Holder) Transactions(ctx context.Context) (map[string]Transaction, error) {
|
||||
return h.transactionManager.List(ctx)
|
||||
}
|
||||
|
||||
func (h *Holder) GetTransaction(id string) (Transaction, error) {
|
||||
return h.transactionManager.Get(id)
|
||||
func (h *Holder) GetTransaction(ctx context.Context, id string) (Transaction, error) {
|
||||
return h.transactionManager.Get(ctx, id)
|
||||
}
|
||||
|
||||
// lockedChan looks a little ridiculous admittedly, but exists for good reason.
|
||||
|
|
|
|||
|
|
@ -1046,7 +1046,7 @@ func (h *Handler) handleGetTransactions(w http.ResponseWriter, r *http.Request)
|
|||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
trnsMap, err := h.api.Transactions()
|
||||
trnsMap, err := h.api.Transactions(r.Context())
|
||||
if err != nil {
|
||||
switch errors.Cause(err) {
|
||||
case pilosa.ErrNodeNotCoordinator:
|
||||
|
|
@ -1107,7 +1107,7 @@ func (h *Handler) handleGetTransaction(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
id := mux.Vars(r)["id"]
|
||||
trns, err := h.api.GetTransaction(id, false)
|
||||
trns, err := h.api.GetTransaction(r.Context(), id, false)
|
||||
h.doTransactionResponse(w, err, trns)
|
||||
}
|
||||
|
||||
|
|
@ -1130,7 +1130,7 @@ func (h *Handler) handlePostTransaction(w http.ResponseWriter, r *http.Request)
|
|||
if !ok {
|
||||
id = reqTrns.ID
|
||||
}
|
||||
trns, err := h.api.StartTransaction(id, reqTrns.Timeout, reqTrns.Exclusive, false)
|
||||
trns, err := h.api.StartTransaction(r.Context(), id, reqTrns.Timeout, reqTrns.Exclusive, false)
|
||||
|
||||
h.doTransactionResponse(w, err, trns)
|
||||
}
|
||||
|
|
@ -1141,7 +1141,7 @@ func (h *Handler) handlePostFinishTransaction(w http.ResponseWriter, r *http.Req
|
|||
return
|
||||
}
|
||||
id := mux.Vars(r)["id"]
|
||||
trns, err := h.api.FinishTransaction(id, false)
|
||||
trns, err := h.api.FinishTransaction(r.Context(), id, false)
|
||||
h.doTransactionResponse(w, err, trns)
|
||||
}
|
||||
|
||||
|
|
|
|||
29
server.go
29
server.go
|
|
@ -808,19 +808,20 @@ func (s *Server) receiveMessage(m Message) error {
|
|||
|
||||
func (s *Server) handleTransactionMessage(tm *TransactionMessage) error {
|
||||
mtrns := tm.Transaction // message transaction
|
||||
ctx := context.Background()
|
||||
switch tm.Action {
|
||||
case TRANSACTION_START:
|
||||
_, err := s.StartTransaction(mtrns.ID, mtrns.Timeout, mtrns.Exclusive, true)
|
||||
_, err := s.StartTransaction(ctx, 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)
|
||||
_, err := s.FinishTransaction(ctx, mtrns.ID, true)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "finishing transaction locally")
|
||||
}
|
||||
case TRANSACTION_VALIDATE:
|
||||
trns, err := s.GetTransaction(mtrns.ID, true)
|
||||
trns, err := s.GetTransaction(ctx, mtrns.ID, true)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "getting local transaction to validate")
|
||||
}
|
||||
|
|
@ -1024,7 +1025,7 @@ func (s *Server) monitorRuntime() {
|
|||
}
|
||||
}
|
||||
|
||||
func (srv *Server) StartTransaction(id string, timeout time.Duration, exclusive bool, remote bool) (Transaction, error) {
|
||||
func (srv *Server) StartTransaction(ctx context.Context, 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
|
||||
|
|
@ -1038,7 +1039,7 @@ func (srv *Server) StartTransaction(id string, timeout time.Duration, exclusive
|
|||
if id == "" {
|
||||
id = uuid.NewV4().String()
|
||||
}
|
||||
trns, err := srv.holder.StartTransaction(id, timeout, exclusive)
|
||||
trns, err := srv.holder.StartTransaction(ctx, id, timeout, exclusive)
|
||||
if err != nil {
|
||||
return trns, errors.Wrap(err, "starting transaction")
|
||||
}
|
||||
|
|
@ -1049,7 +1050,7 @@ func (srv *Server) StartTransaction(id string, timeout time.Duration, exclusive
|
|||
})
|
||||
if err != nil {
|
||||
// try to clean up, but ignore errors
|
||||
srv.holder.FinishTransaction(id)
|
||||
srv.holder.FinishTransaction(ctx, id)
|
||||
srv.SendSync(
|
||||
&TransactionMessage{
|
||||
Action: TRANSACTION_FINISH,
|
||||
|
|
@ -1060,12 +1061,12 @@ func (srv *Server) StartTransaction(id string, timeout time.Duration, exclusive
|
|||
}
|
||||
return trns, nil
|
||||
} else { // remote
|
||||
return srv.holder.StartTransaction(id, timeout, exclusive)
|
||||
return srv.holder.StartTransaction(ctx, id, timeout, exclusive)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (srv *Server) FinishTransaction(id string, remote bool) (Transaction, error) {
|
||||
func (srv *Server) FinishTransaction(ctx context.Context, id string, remote bool) (Transaction, error) {
|
||||
node := srv.node()
|
||||
if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
|
||||
return Transaction{}, ErrNodeNotCoordinator
|
||||
|
|
@ -1075,7 +1076,7 @@ func (srv *Server) FinishTransaction(id string, remote bool) (Transaction, error
|
|||
}
|
||||
|
||||
if !remote {
|
||||
trns, err := srv.holder.FinishTransaction(id)
|
||||
trns, err := srv.holder.FinishTransaction(ctx, id)
|
||||
if err != nil {
|
||||
return trns, errors.Wrap(err, "finishing transaction")
|
||||
}
|
||||
|
|
@ -1091,21 +1092,21 @@ func (srv *Server) FinishTransaction(id string, remote bool) (Transaction, error
|
|||
}
|
||||
return trns, nil
|
||||
} else { // remote
|
||||
return srv.holder.FinishTransaction(id)
|
||||
return srv.holder.FinishTransaction(ctx, id)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (srv *Server) Transactions() (map[string]Transaction, error) {
|
||||
func (srv *Server) Transactions(ctx context.Context) (map[string]Transaction, error) {
|
||||
node := srv.node()
|
||||
if !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
|
||||
return nil, ErrNodeNotCoordinator
|
||||
}
|
||||
|
||||
return srv.holder.Transactions()
|
||||
return srv.holder.Transactions(ctx)
|
||||
}
|
||||
|
||||
func (srv *Server) GetTransaction(id string, remote bool) (Transaction, error) {
|
||||
func (srv *Server) GetTransaction(ctx context.Context, id string, remote bool) (Transaction, error) {
|
||||
node := srv.node()
|
||||
if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 {
|
||||
return Transaction{}, ErrNodeNotCoordinator
|
||||
|
|
@ -1115,7 +1116,7 @@ func (srv *Server) GetTransaction(id string, remote bool) (Transaction, error) {
|
|||
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)
|
||||
trns, err := srv.holder.GetTransaction(ctx, id)
|
||||
if err != nil {
|
||||
return Transaction{}, errors.Wrap(err, "getting transaction")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -380,29 +380,30 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
|
||||
api0 := cluster[0].API
|
||||
api1 := cluster[1].API
|
||||
ctx := context.Background()
|
||||
//api2 := cluster[2].API
|
||||
|
||||
// can fetch empty transactions
|
||||
if trnsMap, err := api0.Transactions(); err != nil {
|
||||
if trnsMap, err := api0.Transactions(ctx); 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 {
|
||||
if _, err := api1.Transactions(ctx); 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 {
|
||||
if trns, err := api0.StartTransaction(ctx, "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 {
|
||||
if trns, err := api1.GetTransaction(ctx, "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)
|
||||
|
|
@ -410,7 +411,7 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
|
||||
// can start transaction with blank id and get uuid back
|
||||
id := ""
|
||||
if trns, err := api0.StartTransaction(id, time.Minute, false, false); err != nil {
|
||||
if trns, err := api0.StartTransaction(ctx, id, time.Minute, false, false); err != nil {
|
||||
t.Errorf("couldn't start transaction: %v", err)
|
||||
} else {
|
||||
id = trns.ID
|
||||
|
|
@ -421,60 +422,60 @@ func TestTransactionsAPI(t *testing.T) {
|
|||
}
|
||||
|
||||
// can't finish transaction on non-coordinator
|
||||
if _, err := api1.FinishTransaction(id, false); err != pilosa.ErrNodeNotCoordinator {
|
||||
if _, err := api1.FinishTransaction(ctx, 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 {
|
||||
if _, err := api0.FinishTransaction(ctx, id, false); err != nil {
|
||||
t.Errorf("couldn't finish transaction: %v", err)
|
||||
}
|
||||
|
||||
// can finish previous transaction
|
||||
if _, err := api0.FinishTransaction("a", false); err != nil {
|
||||
if _, err := api0.FinishTransaction(ctx, "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 {
|
||||
if te, err := api0.StartTransaction(ctx, "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 {
|
||||
if _, err := api0.FinishTransaction(ctx, "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 {
|
||||
if trns, err := api0.StartTransaction(ctx, "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 {
|
||||
if te, err := api0.StartTransaction(ctx, "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 {
|
||||
if _, err := api0.FinishTransaction(ctx, "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 {
|
||||
if trns, err := api0.GetTransaction(ctx, "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 {
|
||||
if trns, err := api1.GetTransaction(ctx, "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)
|
||||
|
|
@ -1167,4 +1168,5 @@ Set("h", adec=100.22)
|
|||
} else if !strings.Contains(result.Body, `"count":1`) {
|
||||
t.Fatalf("expected count 1, but got: '%s'", result.Body)
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package pilosa
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
|
@ -72,7 +73,7 @@ func NewTransactionManager(store TransactionStore) *TransactionManager {
|
|||
// is returned—this is primarily so that the caller can discover if an
|
||||
// exclusive transaction has been made immediately active or if they
|
||||
// need to poll.
|
||||
func (tm *TransactionManager) Start(id string, timeout time.Duration, exclusive bool) (Transaction, error) {
|
||||
func (tm *TransactionManager) Start(ctx context.Context, id string, timeout time.Duration, exclusive bool) (Transaction, error) {
|
||||
tm.mu.Lock()
|
||||
defer tm.mu.Unlock()
|
||||
|
||||
|
|
@ -120,7 +121,7 @@ func (tm *TransactionManager) Start(id string, timeout time.Duration, exclusive
|
|||
|
||||
// Finish completes and removes a transaction, returning the completed
|
||||
// transaction (so that the caller can e.g. view the Stats)
|
||||
func (tm *TransactionManager) Finish(id string) (Transaction, error) {
|
||||
func (tm *TransactionManager) Finish(ctx context.Context, id string) (Transaction, error) {
|
||||
tm.mu.Lock()
|
||||
defer tm.mu.Unlock()
|
||||
return tm.finish(id)
|
||||
|
|
@ -165,7 +166,7 @@ func (tm *TransactionManager) finish(id string) (Transaction, error) {
|
|||
|
||||
// Get retrieves the transaction with the given ID. Returns ErrTransactionNotFound
|
||||
// if there isn't one.
|
||||
func (tm *TransactionManager) Get(id string) (Transaction, error) {
|
||||
func (tm *TransactionManager) Get(ctx context.Context, id string) (Transaction, error) {
|
||||
tm.mu.RLock()
|
||||
defer tm.mu.RUnlock()
|
||||
|
||||
|
|
@ -174,7 +175,7 @@ func (tm *TransactionManager) Get(id string) (Transaction, error) {
|
|||
|
||||
// List returns map of all transactions by their ID. It is a copy and
|
||||
// so may be retained and modified by the caller.
|
||||
func (tm *TransactionManager) List() (map[string]Transaction, error) {
|
||||
func (tm *TransactionManager) List(ctx context.Context) (map[string]Transaction, error) {
|
||||
tm.mu.RLock()
|
||||
defer tm.mu.RUnlock()
|
||||
return tm.store.List()
|
||||
|
|
@ -183,7 +184,7 @@ func (tm *TransactionManager) List() (map[string]Transaction, error) {
|
|||
// ResetDeadline updates the deadline for the transaction with the
|
||||
// given ID to be equal to the current time plus the transaction's
|
||||
// timeout.
|
||||
func (tm *TransactionManager) ResetDeadline(id string) (Transaction, error) {
|
||||
func (tm *TransactionManager) ResetDeadline(ctx context.Context, id string) (Transaction, error) {
|
||||
tm.mu.Lock()
|
||||
defer tm.mu.Unlock()
|
||||
trns, err := tm.store.Get(id)
|
||||
|
|
|
|||
|
|
@ -161,7 +161,9 @@ goes through API (and is passed directly to Server). (unimplemented)
|
|||
|
||||
- [x] implement api layer and cluster logic, startup, etc.
|
||||
- [ ] add new cluster state to explicitly reject certain requests during exclusive transaction?
|
||||
- [ ] implement HTTP layer including header/transaction ID
|
||||
- [x] implement HTTP layer
|
||||
- [x] implement transaction id in header
|
||||
- [x] propagate context
|
||||
- [ ] implement and use persistent transaction store rather than inmem.
|
||||
- [ ] update go-pilosa/gpexp to actually USE transactions
|
||||
- [ ] update IDK to use updated go-pilosa
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package pilosa_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
|
@ -18,6 +19,7 @@ func TestTransactionManager(t *testing.T) {
|
|||
|
||||
tm := pilosa.NewTransactionManager(store)
|
||||
tm.Log = test.NewBufferLogger()
|
||||
ctx := context.Background()
|
||||
|
||||
// can add a non-exclusive transaction
|
||||
trns1 := mustStart(t, tm, "a", time.Microsecond, false)
|
||||
|
|
@ -28,7 +30,7 @@ func TestTransactionManager(t *testing.T) {
|
|||
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)
|
||||
t3, err := tm.Start(ctx, "a", time.Second, true)
|
||||
if err != pilosa.ErrTransactionExists {
|
||||
t.Errorf("expected transaction exists, but got: '%v'", err)
|
||||
}
|
||||
|
|
@ -51,19 +53,19 @@ func TestTransactionManager(t *testing.T) {
|
|||
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 {
|
||||
if _, err := tm.Start(ctx, "d", time.Millisecond, false); err != pilosa.ErrTransactionExclusive {
|
||||
t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err)
|
||||
}
|
||||
|
||||
// can't start new exclusive transactions while an exclusive transaction is pending
|
||||
if _, err := tm.Start("ee", time.Millisecond, true); err != pilosa.ErrTransactionExclusive {
|
||||
if _, err := tm.Start(ctx, "ee", time.Millisecond, true); err != pilosa.ErrTransactionExclusive {
|
||||
t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err)
|
||||
}
|
||||
|
||||
// exclusive transaction becomes active after deadlines expire
|
||||
for i := 0; true; i++ {
|
||||
time.Sleep(time.Microsecond)
|
||||
trnsE, err := tm.Get("ce")
|
||||
trnsE, err := tm.Get(ctx, "ce")
|
||||
if err != nil {
|
||||
t.Errorf("error retrieving exclusive transaction: %v", err)
|
||||
}
|
||||
|
|
@ -76,19 +78,19 @@ func TestTransactionManager(t *testing.T) {
|
|||
}
|
||||
|
||||
// can't start new transactions while an exclusive transaction is active
|
||||
if _, err := tm.Start("f", time.Millisecond, false); err != pilosa.ErrTransactionExclusive {
|
||||
if _, err := tm.Start(ctx, "f", time.Millisecond, false); err != pilosa.ErrTransactionExclusive {
|
||||
t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err)
|
||||
}
|
||||
|
||||
// can't start new exclusive transactions while an exclusive transaction is active
|
||||
if _, err := tm.Start("ge", time.Millisecond, true); err != pilosa.ErrTransactionExclusive {
|
||||
if _, err := tm.Start(ctx, "ge", time.Millisecond, true); err != pilosa.ErrTransactionExclusive {
|
||||
t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err)
|
||||
}
|
||||
|
||||
// exclusive transaction gets expired after other transactions have attempted to start
|
||||
for i := 0; true; i++ {
|
||||
time.Sleep(time.Millisecond * 2)
|
||||
trnsE, err := tm.Get("ce")
|
||||
trnsE, err := tm.Get(ctx, "ce")
|
||||
if err == nil {
|
||||
if i > 10 {
|
||||
t.Fatalf("exclusive transaction didn't expire: %+v", trnsE)
|
||||
|
|
@ -105,7 +107,7 @@ func TestTransactionManager(t *testing.T) {
|
|||
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 {
|
||||
if _, err := tm.Start(ctx, "i", time.Millisecond, false); err != pilosa.ErrTransactionExclusive {
|
||||
t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err)
|
||||
}
|
||||
|
||||
|
|
@ -150,7 +152,7 @@ func TestTransactionManager(t *testing.T) {
|
|||
time.Sleep(time.Millisecond * 3)
|
||||
|
||||
// reset deadline
|
||||
trnsM_reset, err := tm.ResetDeadline("m")
|
||||
trnsM_reset, err := tm.ResetDeadline(ctx, "m")
|
||||
if err != nil {
|
||||
t.Errorf("resetting deadline: %v", err)
|
||||
}
|
||||
|
|
@ -168,7 +170,7 @@ func TestTransactionManager(t *testing.T) {
|
|||
|
||||
func mustStart(t *testing.T, tm *pilosa.TransactionManager, id string, timeout time.Duration, exclusive bool) pilosa.Transaction {
|
||||
t.Helper()
|
||||
trns, err := tm.Start(id, timeout, exclusive)
|
||||
trns, err := tm.Start(context.Background(), id, timeout, exclusive)
|
||||
if err != nil {
|
||||
t.Errorf("starting transaction: %v", err)
|
||||
}
|
||||
|
|
@ -177,7 +179,7 @@ func mustStart(t *testing.T, tm *pilosa.TransactionManager, id string, timeout t
|
|||
|
||||
func mustFinish(t *testing.T, tm *pilosa.TransactionManager, id string) pilosa.Transaction {
|
||||
t.Helper()
|
||||
trns, err := tm.Finish(id)
|
||||
trns, err := tm.Finish(context.Background(), id)
|
||||
if err != nil {
|
||||
t.Errorf("finishing transaction: %v", err)
|
||||
}
|
||||
|
|
@ -186,7 +188,7 @@ func mustFinish(t *testing.T, tm *pilosa.TransactionManager, id string) pilosa.T
|
|||
|
||||
func mustGet(t *testing.T, tm *pilosa.TransactionManager, id string) pilosa.Transaction {
|
||||
t.Helper()
|
||||
trns, err := tm.Get(id)
|
||||
trns, err := tm.Get(context.Background(), id)
|
||||
if err != nil {
|
||||
t.Errorf("getting transaction %s: %v", id, err)
|
||||
}
|
||||
|
|
@ -195,7 +197,7 @@ func mustGet(t *testing.T, tm *pilosa.TransactionManager, id string) pilosa.Tran
|
|||
|
||||
func mustList(t *testing.T, tm *pilosa.TransactionManager) map[string]pilosa.Transaction {
|
||||
t.Helper()
|
||||
trnsMap, err := tm.List()
|
||||
trnsMap, err := tm.List(context.Background())
|
||||
if err != nil {
|
||||
t.Errorf("getting transaction list: %v", err)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue