diff --git a/api.go b/api.go index 6ca4e739a..761a14dc5 100644 --- a/api.go +++ b/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 { diff --git a/holder.go b/holder.go index b345c2a96..15d1dfceb 100644 --- a/holder.go +++ b/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. diff --git a/http/handler.go b/http/handler.go index de17bcea9..2fe22d26d 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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) } diff --git a/server.go b/server.go index cbf580c14..d4f59996c 100644 --- a/server.go +++ b/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") } diff --git a/server/server_test.go b/server/server_test.go index a5b3837f4..2eca8bca9 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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) } + } diff --git a/transaction.go b/transaction.go index ca053fc9c..2d890d905 100644 --- a/transaction.go +++ b/transaction.go @@ -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) diff --git a/transaction.md b/transaction.md index 69ed09fed..389000a23 100644 --- a/transaction.md +++ b/transaction.md @@ -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 diff --git a/transaction_test.go b/transaction_test.go index 917eaf982..2131c6df9 100644 --- a/transaction_test.go +++ b/transaction_test.go @@ -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) }