From 2bdd5a6fec2864ea37f7dace513e3f02771fff8d Mon Sep 17 00:00:00 2001 From: ivcosla Date: Wed, 15 May 2019 19:11:44 +0200 Subject: [PATCH 1/6] added retry on sequence error for update --- pkg/messaging-discovery/client/client.go | 38 +++++++++++++++++++----- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/pkg/messaging-discovery/client/client.go b/pkg/messaging-discovery/client/client.go index cef0331be4..6da9cae166 100644 --- a/pkg/messaging-discovery/client/client.go +++ b/pkg/messaging-discovery/client/client.go @@ -9,6 +9,7 @@ import ( "fmt" "io/ioutil" "net/http" + "sync" "time" "github.com/skycoin/skywire/pkg/cipher" @@ -25,8 +26,9 @@ type APIClient interface { // HTTPClient represents a client that communicates with a messaging-discovery service through http, it // implements APIClient type httpClient struct { - client http.Client - address string + client http.Client + address string + updateMux sync.Mutex // for thread-safe sequence incrementing } // NewHTTP constructs a new APIClient that communicates with discovery via http. @@ -116,6 +118,9 @@ func (c *httpClient) SetEntry(ctx context.Context, e *Entry) error { // UpdateEntry updates Entry in messaging discovery. func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry) error { + c.updateMux.Lock() + defer c.updateMux.Unlock() + e.Sequence++ e.Timestamp = time.Now().UnixNano() err := e.Sign(sk) @@ -123,12 +128,31 @@ func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry return err } - err = c.SetEntry(ctx, e) - if err != nil { - e.Sequence-- + for { + err = c.SetEntry(ctx, e) + if err != nil && err != ErrValidationWrongSequence { + e.Sequence-- + return err + } + if err == ErrValidationWrongSequence { + rE, entryErr := c.Entry(ctx, e.Static) + if entryErr != nil { + return err + } + if rE.Timestamp > e.Timestamp { // If there is a more up to date entry drop update + e.Sequence = rE.Sequence + return nil + } + e.Sequence = rE.Sequence + 1 + err := e.Sign(sk) + if err != nil { + return err + } + } + if err == nil { + return nil + } } - - return err } // AvailableServers returns list of available servers. From 40f71f93aad1f4f651a7a920f9641da6cb0163d2 Mon Sep 17 00:00:00 2001 From: ivcosla Date: Thu, 16 May 2019 21:17:52 +0200 Subject: [PATCH 2/6] bug seems fixed --- pkg/messaging-discovery/client/client.go | 23 ++++++++--------- .../client/client_mock_test.go | 8 ++++++ pkg/messaging-discovery/client/entry.go | 24 ++++++++++++++++++ .../client/http_message.go | 11 ++++++-- pkg/messaging-discovery/client/testing.go | 25 ++++++++++++++++++- 5 files changed, 75 insertions(+), 16 deletions(-) diff --git a/pkg/messaging-discovery/client/client.go b/pkg/messaging-discovery/client/client.go index 6da9cae166..f70068becd 100644 --- a/pkg/messaging-discovery/client/client.go +++ b/pkg/messaging-discovery/client/client.go @@ -5,7 +5,6 @@ import ( "bytes" "context" "encoding/json" - "errors" "fmt" "io/ioutil" "net/http" @@ -65,7 +64,7 @@ func (c *httpClient) Entry(ctx context.Context, publicKey cipher.PubKey) (*Entry return nil, err } - return nil, errors.New(message.String()) + return nil, errFromString(message.String()) } err = json.NewDecoder(resp.Body).Decode(&entry) @@ -110,8 +109,7 @@ func (c *httpClient) SetEntry(ctx context.Context, e *Entry) error { if err != nil { return err } - - return errors.New(httpResponse.String()) + return errFromString(httpResponse.Message) } return nil } @@ -130,11 +128,11 @@ func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry for { err = c.SetEntry(ctx, e) - if err != nil && err != ErrValidationWrongSequence { - e.Sequence-- - return err - } - if err == ErrValidationWrongSequence { + if err != nil { + if err != ErrValidationWrongSequence { + e.Sequence-- + return err + } rE, entryErr := c.Entry(ctx, e.Static) if entryErr != nil { return err @@ -148,10 +146,9 @@ func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry if err != nil { return err } + continue } - if err == nil { - return nil - } + return nil } } @@ -181,7 +178,7 @@ func (c *httpClient) AvailableServers(ctx context.Context) ([]*Entry, error) { return nil, err } - return nil, errors.New(message.String()) + return nil, errFromString(message.String()) } err = json.NewDecoder(resp.Body).Decode(&entries) diff --git a/pkg/messaging-discovery/client/client_mock_test.go b/pkg/messaging-discovery/client/client_mock_test.go index f407b86009..23a0ee4284 100644 --- a/pkg/messaging-discovery/client/client_mock_test.go +++ b/pkg/messaging-discovery/client/client_mock_test.go @@ -287,6 +287,14 @@ func TestNewMockUpdateEntriesEndpoint(t *testing.T) { e.Server.Address = "different one" }, }, + { + name: "udpate retries on wrong sequence", + responseShouldError: false, + secretKey: sk, + entryPreHook: func(entry *client.Entry) { + entry.Sequence = 3 + }, + }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/pkg/messaging-discovery/client/entry.go b/pkg/messaging-discovery/client/entry.go index 73bd79a0b0..660e2c3cdd 100644 --- a/pkg/messaging-discovery/client/entry.go +++ b/pkg/messaging-discovery/client/entry.go @@ -28,8 +28,32 @@ var ( ErrValidationNoClientOrServer = NewEntryValidationError("entry has neither client or server field") ErrValidationWrongSequence = NewEntryValidationError("sequence field of new entry is not sequence of old entry + 1") ErrValidationWrongTime = NewEntryValidationError("previous entry timestamp is not set before current entry timestamp") + + errReverseMap = map[string]error{ + ErrKeyNotFound.Error(): ErrKeyNotFound, + ErrUnexpected.Error(): ErrUnexpected, + ErrUnauthorized.Error(): ErrUnauthorized, + ErrBadInput.Error(): ErrBadInput, + ErrValidationNonZeroSequence.Error(): ErrValidationNonZeroSequence, + ErrValidationNilEphemerals.Error(): ErrValidationNilEphemerals, + ErrValidationNilKeys.Error(): ErrValidationNilKeys, + ErrValidationNonNilEphemerals.Error(): ErrValidationNonNilEphemerals, + ErrValidationNoSignature.Error(): ErrValidationNoSignature, + ErrValidationNoVersion.Error(): ErrValidationNoVersion, + ErrValidationNoClientOrServer.Error(): ErrValidationNoClientOrServer, + ErrValidationWrongSequence.Error(): ErrValidationWrongSequence, + ErrValidationWrongTime.Error(): ErrValidationWrongTime, + } ) +func errFromString(s string) error { + err, ok := errReverseMap[s] + if !ok { + return ErrUnexpected + } + return err +} + // EntryValidationError represents transient error caused by invalid // data in Entry type EntryValidationError struct { diff --git a/pkg/messaging-discovery/client/http_message.go b/pkg/messaging-discovery/client/http_message.go index 6d50604a85..f5adc68bef 100644 --- a/pkg/messaging-discovery/client/http_message.go +++ b/pkg/messaging-discovery/client/http_message.go @@ -1,14 +1,15 @@ package client import ( + "errors" "fmt" "net/http" ) // Exposed Http messages var ( - MsgEntrySet = HTTPMessage{Code: http.StatusOK, Message: "wrote a new entry"} - MsgEntryUpdated = HTTPMessage{Code: http.StatusOK, Message: "wrote new entry iteration"} + MsgEntrySet = HTTPMessage{Code: http.StatusOK, Message: "wrote a new entry"} + MsgEntryUpdated = HTTPMessage{Code: http.StatusOK, Message: "wrote new entry iteration"} ) // HTTPMessage represents a message to be returned as an http response @@ -20,3 +21,9 @@ type HTTPMessage struct { func (h HTTPMessage) String() string { return fmt.Sprintf("status code: %d. message: %s", h.Code, h.Message) } + +// ToError returns an error representing the httpMessage for error comparisons, this is preferred for this type instead +// of implementing error +func (h HTTPMessage) ToError() error { + return errors.New(h.String()) +} diff --git a/pkg/messaging-discovery/client/testing.go b/pkg/messaging-discovery/client/testing.go index 8fd77ade46..c0483d995b 100644 --- a/pkg/messaging-discovery/client/testing.go +++ b/pkg/messaging-discovery/client/testing.go @@ -91,7 +91,30 @@ func (m *mockClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry return err } - return m.SetEntry(ctx, e) + for { + err = m.SetEntry(ctx, e) + if err != nil { + if err != ErrValidationWrongSequence { + e.Sequence-- + return err + } + rE, entryErr := m.Entry(ctx, e.Static) + if entryErr != nil { + return err + } + if rE.Timestamp > e.Timestamp { // If there is a more up to date entry drop update + e.Sequence = rE.Sequence + return nil + } + e.Sequence = rE.Sequence + 1 + err := e.Sign(sk) + if err != nil { + return err + } + continue + } + return nil + } } // AvailableServers returns all the servers that the APIClient mock has From 6efde168f8da9355daaead3394a9fa20fcd65038 Mon Sep 17 00:00:00 2001 From: ivcosla Date: Fri, 17 May 2019 09:49:14 +0200 Subject: [PATCH 3/6] format --- pkg/messaging-discovery/client/http_message.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/messaging-discovery/client/http_message.go b/pkg/messaging-discovery/client/http_message.go index f5adc68bef..39b2a0bd3f 100644 --- a/pkg/messaging-discovery/client/http_message.go +++ b/pkg/messaging-discovery/client/http_message.go @@ -8,8 +8,8 @@ import ( // Exposed Http messages var ( - MsgEntrySet = HTTPMessage{Code: http.StatusOK, Message: "wrote a new entry"} - MsgEntryUpdated = HTTPMessage{Code: http.StatusOK, Message: "wrote new entry iteration"} + MsgEntrySet = HTTPMessage{Code: http.StatusOK, Message: "wrote a new entry"} + MsgEntryUpdated = HTTPMessage{Code: http.StatusOK, Message: "wrote new entry iteration"} ) // HTTPMessage represents a message to be returned as an http response From bebafeb86fa925fcf8fbdfd475ce0441176a9c85 Mon Sep 17 00:00:00 2001 From: ivcosla Date: Mon, 20 May 2019 15:43:11 +0200 Subject: [PATCH 4/6] refactored retry --- pkg/messaging-discovery/client/client.go | 37 ++++++++++++------------ 1 file changed, 18 insertions(+), 19 deletions(-) diff --git a/pkg/messaging-discovery/client/client.go b/pkg/messaging-discovery/client/client.go index f70068becd..72d644ae0b 100644 --- a/pkg/messaging-discovery/client/client.go +++ b/pkg/messaging-discovery/client/client.go @@ -128,27 +128,26 @@ func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry for { err = c.SetEntry(ctx, e) + if err == nil { + return nil + } + if err != ErrValidationWrongSequence { + e.Sequence-- + return err + } + rE, entryErr := c.Entry(ctx, e.Static) + if entryErr != nil { + return err + } + if rE.Timestamp > e.Timestamp { // If there is a more up to date entry drop update + e.Sequence = rE.Sequence + return nil + } + e.Sequence = rE.Sequence + 1 + err := e.Sign(sk) if err != nil { - if err != ErrValidationWrongSequence { - e.Sequence-- - return err - } - rE, entryErr := c.Entry(ctx, e.Static) - if entryErr != nil { - return err - } - if rE.Timestamp > e.Timestamp { // If there is a more up to date entry drop update - e.Sequence = rE.Sequence - return nil - } - e.Sequence = rE.Sequence + 1 - err := e.Sign(sk) - if err != nil { - return err - } - continue + return err } - return nil } } From 43197f96debe36ae3619534444cfe0b5d86c806b Mon Sep 17 00:00:00 2001 From: ivcosla Date: Mon, 20 May 2019 15:55:14 +0200 Subject: [PATCH 5/6] refactored testing, removed HttpMessage.ToError() --- .../client/http_message.go | 7 ---- pkg/messaging-discovery/client/testing.go | 37 +++++++++---------- 2 files changed, 18 insertions(+), 26 deletions(-) diff --git a/pkg/messaging-discovery/client/http_message.go b/pkg/messaging-discovery/client/http_message.go index 39b2a0bd3f..6d50604a85 100644 --- a/pkg/messaging-discovery/client/http_message.go +++ b/pkg/messaging-discovery/client/http_message.go @@ -1,7 +1,6 @@ package client import ( - "errors" "fmt" "net/http" ) @@ -21,9 +20,3 @@ type HTTPMessage struct { func (h HTTPMessage) String() string { return fmt.Sprintf("status code: %d. message: %s", h.Code, h.Message) } - -// ToError returns an error representing the httpMessage for error comparisons, this is preferred for this type instead -// of implementing error -func (h HTTPMessage) ToError() error { - return errors.New(h.String()) -} diff --git a/pkg/messaging-discovery/client/testing.go b/pkg/messaging-discovery/client/testing.go index c0483d995b..5c4ca288e6 100644 --- a/pkg/messaging-discovery/client/testing.go +++ b/pkg/messaging-discovery/client/testing.go @@ -93,27 +93,26 @@ func (m *mockClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry for { err = m.SetEntry(ctx, e) + if err == nil { + return nil + } + if err != ErrValidationWrongSequence { + e.Sequence-- + return err + } + rE, entryErr := m.Entry(ctx, e.Static) + if entryErr != nil { + return err + } + if rE.Timestamp > e.Timestamp { // If there is a more up to date entry drop update + e.Sequence = rE.Sequence + return nil + } + e.Sequence = rE.Sequence + 1 + err := e.Sign(sk) if err != nil { - if err != ErrValidationWrongSequence { - e.Sequence-- - return err - } - rE, entryErr := m.Entry(ctx, e.Static) - if entryErr != nil { - return err - } - if rE.Timestamp > e.Timestamp { // If there is a more up to date entry drop update - e.Sequence = rE.Sequence - return nil - } - e.Sequence = rE.Sequence + 1 - err := e.Sign(sk) - if err != nil { - return err - } - continue + return err } - return nil } } From 90b8f84db486a953afd5e22d3f32ffe110488c78 Mon Sep 17 00:00:00 2001 From: ivcosla Date: Tue, 21 May 2019 16:14:58 +0200 Subject: [PATCH 6/6] sign moved to begining of loop --- pkg/messaging-discovery/client/client.go | 12 ++++-------- pkg/messaging-discovery/client/testing.go | 12 ++++-------- 2 files changed, 8 insertions(+), 16 deletions(-) diff --git a/pkg/messaging-discovery/client/client.go b/pkg/messaging-discovery/client/client.go index 72d644ae0b..fd063c9974 100644 --- a/pkg/messaging-discovery/client/client.go +++ b/pkg/messaging-discovery/client/client.go @@ -121,12 +121,12 @@ func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry e.Sequence++ e.Timestamp = time.Now().UnixNano() - err := e.Sign(sk) - if err != nil { - return err - } for { + err := e.Sign(sk) + if err != nil { + return err + } err = c.SetEntry(ctx, e) if err == nil { return nil @@ -144,10 +144,6 @@ func (c *httpClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry return nil } e.Sequence = rE.Sequence + 1 - err := e.Sign(sk) - if err != nil { - return err - } } } diff --git a/pkg/messaging-discovery/client/testing.go b/pkg/messaging-discovery/client/testing.go index 5c4ca288e6..e9327dee6c 100644 --- a/pkg/messaging-discovery/client/testing.go +++ b/pkg/messaging-discovery/client/testing.go @@ -86,12 +86,12 @@ func (m *mockClient) SetEntry(ctx context.Context, e *Entry) error { func (m *mockClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry) error { e.Sequence++ e.Timestamp = time.Now().UnixNano() - err := e.Sign(sk) - if err != nil { - return err - } for { + err := e.Sign(sk) + if err != nil { + return err + } err = m.SetEntry(ctx, e) if err == nil { return nil @@ -109,10 +109,6 @@ func (m *mockClient) UpdateEntry(ctx context.Context, sk cipher.SecKey, e *Entry return nil } e.Sequence = rE.Sequence + 1 - err := e.Sign(sk) - if err != nil { - return err - } } }