From d82b2dad92e386b9a1d53414e740a843b085b3d0 Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Tue, 23 Mar 2021 10:00:56 -0400
Subject: [PATCH 01/21] feat: cmd: implement import-runtime subcommand (#1483)
---
cmd/gossamer/import_runtime.go | 48 +++++++++++++++++++++++++++
cmd/gossamer/import_runtime_test.go | 50 +++++++++++++++++++++++++++++
cmd/gossamer/main.go | 34 ++++++++++----------
lib/genesis/genesis.go | 2 +-
lib/genesis/helpers.go | 34 +++++++++++++-------
5 files changed, 138 insertions(+), 30 deletions(-)
create mode 100644 cmd/gossamer/import_runtime.go
create mode 100644 cmd/gossamer/import_runtime_test.go
diff --git a/cmd/gossamer/import_runtime.go b/cmd/gossamer/import_runtime.go
new file mode 100644
index 0000000000..bc1b53a68a
--- /dev/null
+++ b/cmd/gossamer/import_runtime.go
@@ -0,0 +1,48 @@
+// Copyright 2019 ChainSafe Systems (ON) Corp.
+// This file is part of gossamer.
+//
+// The gossamer library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The gossamer library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the gossamer library. If not, see .
+
+package main
+
+import (
+ "encoding/json"
+ "fmt"
+ "io/ioutil"
+ "path/filepath"
+
+ "github.com/ChainSafe/gossamer/lib/genesis"
+)
+
+var defaultGenesisSpecPath = "./chain/gssmr/genesis.json"
+
+func createGenesisWithRuntime(fp string) (string, error) {
+ runtime, err := ioutil.ReadFile(filepath.Clean(fp))
+ if err != nil {
+ return "", err
+ }
+
+ genesis, err := genesis.NewGenesisSpecFromJSON(defaultGenesisSpecPath)
+ if err != nil {
+ return "", err
+ }
+
+ genesis.Genesis.Runtime["system"]["code"] = fmt.Sprintf("0x%x", runtime)
+ bz, err := json.MarshalIndent(genesis, "", "\t")
+ if err != nil {
+ return "", err
+ }
+
+ return string(bz), nil
+}
diff --git a/cmd/gossamer/import_runtime_test.go b/cmd/gossamer/import_runtime_test.go
new file mode 100644
index 0000000000..c1434a9fe6
--- /dev/null
+++ b/cmd/gossamer/import_runtime_test.go
@@ -0,0 +1,50 @@
+// Copyright 2019 ChainSafe Systems (ON) Corp.
+// This file is part of gossamer.
+//
+// The gossamer library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The gossamer library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the gossamer library. If not, see .
+
+package main
+
+import (
+ "encoding/json"
+ "io/ioutil"
+ "os"
+ "testing"
+
+ "github.com/ChainSafe/gossamer/lib/common"
+ "github.com/ChainSafe/gossamer/lib/genesis"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestCreateGenesisWithRuntime(t *testing.T) {
+ defaultGenesisSpecPath = "../../chain/gssmr/genesis.json"
+
+ testCode := []byte("somecode")
+ testHex := common.BytesToHex(testCode)
+ testFile, err := ioutil.TempFile("", "testcode-*.wasm")
+ require.NoError(t, err)
+ defer os.Remove(testFile.Name())
+
+ err = ioutil.WriteFile(testFile.Name(), testCode, 0777)
+ require.NoError(t, err)
+
+ out, err := createGenesisWithRuntime(testFile.Name())
+ require.NoError(t, err)
+
+ g := new(genesis.Genesis)
+ err = json.Unmarshal([]byte(out), g)
+ require.NoError(t, err)
+ require.Equal(t, testHex, g.Genesis.Runtime["system"]["code"].(string))
+}
diff --git a/cmd/gossamer/main.go b/cmd/gossamer/main.go
index e22b432e8d..101027b676 100644
--- a/cmd/gossamer/main.go
+++ b/cmd/gossamer/main.go
@@ -19,9 +19,7 @@ package main
import (
"errors"
"fmt"
- "io/ioutil"
"os"
- "path/filepath"
"github.com/ChainSafe/gossamer/dot"
"github.com/ChainSafe/gossamer/lib/keystore"
@@ -71,7 +69,7 @@ var (
"\tTo import a keystore file: gossamer account --import=path/to/file\n" +
"\tTo list keys: gossamer account --list",
}
- // initCommand defines the "init" subcommand (ie, `gossamer init`)
+ // buildSpecCommand creates a raw genesis file from a human readable genesis file.
buildSpecCommand = cli.Command{
Action: FixFlagOrder(buildSpecAction),
Name: "build-spec",
@@ -81,18 +79,20 @@ var (
Category: "BUILD-SPEC",
Description: "The build-spec command outputs current genesis JSON data.\n" +
"\tUsage: gossamer build-spec\n" +
- "\tTo generate raw genesis file: gossamer build-spec --raw",
+ "\tTo generate raw genesis file from default: gossamer build-spec --raw > genesis-raw.json" +
+ "\tTo generate raw genesis file from specific genesis file: gossamer build-spec --raw --genesis genesis.json > genesis-raw.json",
}
- // TODO: update this to put the wasm into a genesis file
- wasmToHexCommand = cli.Command{
- Action: FixFlagOrder(wasmToHexAction),
- Name: "convert-wasm",
- Usage: "Converts a .wasm file to a hex string to be used in a genesis file",
+
+ // importRuntime generates a genesis file given a .wasm runtime binary.
+ importRuntimeCommand = cli.Command{
+ Action: FixFlagOrder(importRuntimeAction),
+ Name: "import-runtime",
+ Usage: "Generates a genesis file given a .wasm runtime binary",
ArgsUsage: "",
Flags: RootFlags,
- Category: "CONVERT-WASM",
- Description: "The convert-wasm command converts a .wasm file to a hex string to be used in a genesis file.\n" +
- "\tUsage: gossamer convert-wasm runtime.wasm\n",
+ Category: "IMPORT-RUNTIME",
+ Description: "The import-runtime command generates a genesis file given a .wasm runtime binary.\n" +
+ "\tUsage: gossamer import-runtime runtime.wasm > genesis.json\n",
}
importStateCommand = cli.Command{
@@ -121,7 +121,7 @@ func init() {
initCommand,
accountCommand,
buildSpecCommand,
- wasmToHexCommand,
+ importRuntimeCommand,
importStateCommand,
}
app.Flags = RootFlags
@@ -163,20 +163,20 @@ func importStateAction(ctx *cli.Context) error {
return dot.ImportState(cfg.Global.BasePath, stateFP, headerFP, uint64(firstSlot))
}
-// wasmToHexAction converts a .wasm file to a hex string and outputs it to stdout
-func wasmToHexAction(ctx *cli.Context) error {
+// importRuntimeAction generates a genesis file given a .wasm runtime binary.
+func importRuntimeAction(ctx *cli.Context) error {
arguments := ctx.Args()
if len(arguments) == 0 {
return fmt.Errorf("no args provided, please provide wasm file")
}
fp := arguments[0]
- bytes, err := ioutil.ReadFile(filepath.Clean(fp))
+ out, err := createGenesisWithRuntime(fp)
if err != nil {
return err
}
- fmt.Printf("0x%x", bytes)
+ fmt.Println(out)
return nil
}
diff --git a/lib/genesis/genesis.go b/lib/genesis/genesis.go
index 13fea973e5..2f0355f9f8 100644
--- a/lib/genesis/genesis.go
+++ b/lib/genesis/genesis.go
@@ -49,7 +49,7 @@ type Data struct {
// Fields stores genesis raw data, and human readable runtime data
type Fields struct {
- Raw map[string]map[string]string `json:"raw"`
+ Raw map[string]map[string]string `json:"raw,omitempty"`
Runtime map[string]map[string]interface{} `json:"runtime,omitempty"`
}
diff --git a/lib/genesis/helpers.go b/lib/genesis/helpers.go
index b47842b12e..ea52aa5ee3 100644
--- a/lib/genesis/helpers.go
+++ b/lib/genesis/helpers.go
@@ -111,37 +111,47 @@ func trimGenesisAuthority(g *Genesis, authCount int) {
// NewGenesisFromJSON parses Human Readable JSON formatted genesis file.Name. If authCount > 0,
// then it keeps only `authCount` number of authorities for babe and grandpa.
func NewGenesisFromJSON(file string, authCount int) (*Genesis, error) {
- fp, err := filepath.Abs(file)
+ g, err := NewGenesisSpecFromJSON(file)
if err != nil {
return nil, err
}
- data, err := ioutil.ReadFile(filepath.Clean(fp))
+ if authCount > 0 {
+ trimGenesisAuthority(g, authCount)
+ }
+
+ grt := g.Genesis.Runtime
+ res, err := buildRawMap(grt)
if err != nil {
return nil, err
}
- g := new(Genesis)
+ g.Genesis.Raw = make(map[string]map[string]string)
+ g.Genesis.Raw["top"] = res
- err = json.Unmarshal(data, g)
+ return g, err
+}
+
+// NewGenesisSpecFromJSON returns a new Genesis (without raw fields) from a human-readable genesis file
+func NewGenesisSpecFromJSON(file string) (*Genesis, error) {
+ fp, err := filepath.Abs(file)
if err != nil {
return nil, err
}
- if authCount > 0 {
- trimGenesisAuthority(g, authCount)
+ data, err := ioutil.ReadFile(filepath.Clean(fp))
+ if err != nil {
+ return nil, err
}
- grt := g.Genesis.Runtime
- res, err := buildRawMap(grt)
+ g := new(Genesis)
+
+ err = json.Unmarshal(data, g)
if err != nil {
return nil, err
}
- g.Genesis.Raw = make(map[string]map[string]string)
- g.Genesis.Raw["top"] = res
-
- return g, err
+ return g, nil
}
// keyValue struct to hold data regarding entry
From dfb95d2b0a5b9bae6bebbe97755ac25915bda7cf Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Wed, 24 Mar 2021 10:29:31 -0400
Subject: [PATCH 02/21] fix(lib/crypto/ed25519): update ed25519 to use
go-schnorrkel bip39 derivation (#1488)
---
lib/crypto/ed25519/ed25519.go | 7 +++++--
lib/crypto/ed25519/ed25519_test.go | 11 +++++++++++
lib/runtime/wasmer/imports.go | 10 +++++-----
lib/runtime/wasmer/imports_test.go | 2 --
4 files changed, 21 insertions(+), 9 deletions(-)
diff --git a/lib/crypto/ed25519/ed25519.go b/lib/crypto/ed25519/ed25519.go
index a1825d196b..196e688d39 100644
--- a/lib/crypto/ed25519/ed25519.go
+++ b/lib/crypto/ed25519/ed25519.go
@@ -27,7 +27,7 @@ import (
"github.com/ChainSafe/gossamer/lib/common"
"github.com/ChainSafe/gossamer/lib/crypto"
- bip39 "github.com/cosmos/go-bip39"
+ "github.com/ChainSafe/go-schnorrkel"
)
// PublicKeyLength is the fixed Public Key Length
@@ -120,7 +120,10 @@ func NewKeypairFromPrivateKeyString(in string) (*Keypair, error) {
// NewKeypairFromMnenomic returns a new Keypair using the given mnemonic and password.
func NewKeypairFromMnenomic(mnemonic, password string) (*Keypair, error) {
- seed := bip39.NewSeed(mnemonic, password)
+ seed, err := schnorrkel.SeedFromMnemonic(mnemonic, password)
+ if err != nil {
+ return nil, err
+ }
return NewKeypairFromSeed(seed[:32])
}
diff --git a/lib/crypto/ed25519/ed25519_test.go b/lib/crypto/ed25519/ed25519_test.go
index 5dbe331900..be44308d9e 100644
--- a/lib/crypto/ed25519/ed25519_test.go
+++ b/lib/crypto/ed25519/ed25519_test.go
@@ -21,6 +21,8 @@ import (
"reflect"
"testing"
+ "github.com/ChainSafe/gossamer/lib/common"
+
bip39 "github.com/cosmos/go-bip39"
"github.com/stretchr/testify/require"
)
@@ -101,3 +103,12 @@ func TestNewKeypairFromMnenomic(t *testing.T) {
_, err = NewKeypairFromMnenomic(mnemonic, "")
require.NoError(t, err)
}
+
+func TestNewKeypairFromMnenomic_Again(t *testing.T) {
+ mnemonic := "twist sausage october vivid neglect swear crumble hawk beauty fabric egg fragile"
+ kp, err := NewKeypairFromMnenomic(mnemonic, "")
+ require.NoError(t, err)
+
+ expectedPubkey := common.MustHexToBytes("0xf56d9231e7b7badd3f1e10ad15ef8aa08b70839723d0a2d10d7329f0ea2b8c61")
+ require.Equal(t, expectedPubkey, kp.Public().Encode())
+}
diff --git a/lib/runtime/wasmer/imports.go b/lib/runtime/wasmer/imports.go
index 1b7c583191..ca4a7fb42e 100644
--- a/lib/runtime/wasmer/imports.go
+++ b/lib/runtime/wasmer/imports.go
@@ -127,8 +127,8 @@ func ext_logging_log_version_1(context unsafe.Pointer, level C.int32_t, targetDa
logger.Trace("[ext_logging_log_version_1] executing...")
instanceContext := wasm.IntoInstanceContext(context)
- target := fmt.Sprintf("%s", asMemorySlice(instanceContext, targetData))
- msg := fmt.Sprintf("%s", asMemorySlice(instanceContext, msgData))
+ target := string(asMemorySlice(instanceContext, targetData))
+ msg := string(asMemorySlice(instanceContext, msgData))
switch int(level) {
case 0:
@@ -215,7 +215,7 @@ func ext_crypto_ed25519_generate_version_1(context unsafe.Pointer, keyTypeID C.i
var kp crypto.Keypair
if seed.Exists() {
- kp, err = ed25519.NewKeypairFromMnenomic(string(seedBytes), "")
+ kp, err = ed25519.NewKeypairFromMnenomic(string(seed.Value()), "")
} else {
kp, err = ed25519.GenerateKeypair()
}
@@ -849,7 +849,7 @@ func ext_misc_print_utf8_version_1(context unsafe.Pointer, dataSpan C.int64_t) {
instanceContext := wasm.IntoInstanceContext(context)
data := asMemorySlice(instanceContext, dataSpan)
- logger.Debug("[ext_misc_print_utf8_version_1]", "utf8", fmt.Sprintf("%s", data))
+ logger.Debug("[ext_misc_print_utf8_version_1]", "utf8", string(data))
}
//export ext_misc_runtime_version_version_1
@@ -1241,7 +1241,7 @@ func ext_hashing_twox_128_version_1(context unsafe.Pointer, dataSpan C.int64_t)
return 0
}
- logger.Debug("[ext_hashing_twox_128_version_1]", "data", fmt.Sprintf("%s", data), "hash", fmt.Sprintf("0x%x", hash))
+ logger.Debug("[ext_hashing_twox_128_version_1]", "data", string(data), "hash", fmt.Sprintf("0x%x", hash))
out, err := toWasmMemorySized(instanceContext, hash, 16)
if err != nil {
diff --git a/lib/runtime/wasmer/imports_test.go b/lib/runtime/wasmer/imports_test.go
index e7cb7c3e96..aba4174383 100644
--- a/lib/runtime/wasmer/imports_test.go
+++ b/lib/runtime/wasmer/imports_test.go
@@ -419,8 +419,6 @@ func Test_ext_crypto_ed25519_generate_version_1(t *testing.T) {
mnemonic, err := crypto.NewBIP39Mnemonic()
require.NoError(t, err)
- // TODO: we currently don't provide a seed since the spec says the seed is an optional BIP-39 seed
- // clarify whether this is a mnemonic or not
data := optional.NewBytes(true, []byte(mnemonic))
seedData, err := data.Encode()
require.NoError(t, err)
From 4529aee09d0eda7792ec90239703e80fad5293f2 Mon Sep 17 00:00:00 2001
From: Arijit Das
Date: Thu, 25 Mar 2021 03:13:41 +0530
Subject: [PATCH 03/21] fix(state) : Update StorageState to load storage from
database. (#1486)
---
dot/state/storage.go | 102 ++++++++++++++++++++------------------
dot/state/storage_test.go | 51 +++++++++++++++++++
lib/trie/child_storage.go | 8 +--
lib/trie/trie.go | 6 +--
4 files changed, 112 insertions(+), 55 deletions(-)
diff --git a/dot/state/storage.go b/dot/state/storage.go
index 01ffc37821..5c9a45be54 100644
--- a/dot/state/storage.go
+++ b/dot/state/storage.go
@@ -204,19 +204,9 @@ func (s *StorageState) LoadFromDB(root common.Hash) (*trie.Trie, error) {
// ExistsStorage check if the key exists in the storage trie with the given storage hash
// If no hash is provided, the current chain head is used
-func (s *StorageState) ExistsStorage(hash *common.Hash, key []byte) (bool, error) {
- if hash == nil {
- sr, err := s.blockState.BestBlockStateRoot()
- if err != nil {
- return false, err
- }
- hash = &sr
- }
-
- s.lock.RLock()
- defer s.lock.RUnlock()
- val := s.tries[*hash].Get(key)
- return val != nil, nil
+func (s *StorageState) ExistsStorage(root *common.Hash, key []byte) (bool, error) {
+ val, err := s.GetStorage(root, key)
+ return val != nil, err
}
// GetStorage gets the object from the trie using the given key and storage hash
@@ -233,8 +223,8 @@ func (s *StorageState) GetStorage(root *common.Hash, key []byte) ([]byte, error)
s.lock.RLock()
defer s.lock.RUnlock()
- if s.tries[*root] != nil {
- val := s.tries[*root].Get(key)
+ if trie, ok := s.tries[*root]; ok {
+ val := trie.Get(key)
return val, nil
}
@@ -262,19 +252,7 @@ func (s *StorageState) GetStateRootFromBlock(bhash *common.Hash) (*common.Hash,
// StorageRoot returns the root hash of the current storage trie
func (s *StorageState) StorageRoot() (common.Hash, error) {
- sr, err := s.blockState.BestBlockStateRoot()
- if err != nil {
- return common.Hash{}, err
- }
-
- s.lock.RLock()
- defer s.lock.RUnlock()
-
- if s.tries[sr] == nil {
- return common.Hash{}, errTrieDoesNotExist(sr)
- }
-
- return s.tries[sr].Hash()
+ return s.blockState.BestBlockStateRoot()
}
// EnumeratedTrieRoot not implemented
@@ -294,17 +272,19 @@ func (s *StorageState) Entries(root *common.Hash) (map[string][]byte, error) {
}
s.lock.RLock()
- defer s.lock.RUnlock()
-
- if s.tries[*root] != nil {
- return s.tries[*root].Entries(), nil
- }
+ tr, ok := s.tries[*root]
+ s.lock.RUnlock()
- tr, err := s.LoadFromDB(*root)
- if err != nil {
- return nil, err
+ if !ok {
+ var err error
+ tr, err = s.LoadFromDB(*root)
+ if err != nil {
+ return nil, errTrieDoesNotExist(*root)
+ }
}
+ s.lock.RLock()
+ defer s.lock.RUnlock()
return tr.Entries(), nil
}
@@ -317,11 +297,22 @@ func (s *StorageState) GetKeysWithPrefix(hash *common.Hash, prefix []byte) ([][]
}
hash = &sr
}
- t := s.tries[*hash]
- if t == nil {
- return nil, fmt.Errorf("unable to retrieve trie with hash %x", *hash)
+
+ s.lock.RLock()
+ tr, ok := s.tries[*hash]
+ s.lock.RUnlock()
+
+ if !ok {
+ var err error
+ tr, err = s.LoadFromDB(*hash)
+ if err != nil {
+ return nil, errTrieDoesNotExist(*hash)
+ }
}
- return t.GetKeysWithPrefix(prefix), nil
+
+ s.lock.RLock()
+ defer s.lock.RUnlock()
+ return tr.GetKeysWithPrefix(prefix), nil
}
// GetStorageChild return GetChild from the trie
@@ -335,13 +326,20 @@ func (s *StorageState) GetStorageChild(hash *common.Hash, keyToChild []byte) (*t
}
s.lock.RLock()
- defer s.lock.RUnlock()
+ tr, ok := s.tries[*hash]
+ s.lock.RUnlock()
- if s.tries[*hash] == nil {
- return nil, errTrieDoesNotExist(*hash)
+ if !ok {
+ var err error
+ tr, err = s.LoadFromDB(*hash)
+ if err != nil {
+ return nil, errTrieDoesNotExist(*hash)
+ }
}
- return s.tries[*hash].GetChild(keyToChild)
+ s.lock.RLock()
+ defer s.lock.RUnlock()
+ return tr.GetChild(keyToChild)
}
// GetStorageFromChild return GetFromChild from the trie
@@ -355,12 +353,20 @@ func (s *StorageState) GetStorageFromChild(hash *common.Hash, keyToChild, key []
}
s.lock.RLock()
- defer s.lock.RUnlock()
+ tr, ok := s.tries[*hash]
+ s.lock.RUnlock()
- if s.tries[*hash] == nil {
- return nil, errTrieDoesNotExist(*hash)
+ if !ok {
+ var err error
+ tr, err = s.LoadFromDB(*hash)
+ if err != nil {
+ return nil, errTrieDoesNotExist(*hash)
+ }
}
- return s.tries[*hash].GetFromChild(keyToChild, key)
+
+ s.lock.RLock()
+ defer s.lock.RUnlock()
+ return tr.GetFromChild(keyToChild, key)
}
// LoadCode returns the runtime code (located at :code)
diff --git a/dot/state/storage_test.go b/dot/state/storage_test.go
index 93b4ca4e7e..5c55c4e5c8 100644
--- a/dot/state/storage_test.go
+++ b/dot/state/storage_test.go
@@ -69,3 +69,54 @@ func TestStorage_GetStorageByBlockHash(t *testing.T) {
require.NoError(t, err)
require.Equal(t, value, res)
}
+
+func TestStorage_LoadFromDB(t *testing.T) {
+ storage := newTestStorageState(t)
+ ts, err := storage.TrieState(&trie.EmptyHash)
+ require.NoError(t, err)
+
+ trieKV := []struct {
+ key []byte
+ value []byte
+ }{{},
+ {[]byte("key1"), []byte("value1")},
+ {[]byte("key2"), []byte("value2")},
+ {[]byte("xyzKey1"), []byte("xyzValue1")},
+ }
+
+ for _, kv := range trieKV {
+ ts.Set(kv.key, kv.value)
+ }
+
+ root, err := ts.Root()
+ require.NoError(t, err)
+
+ // Write trie to disk.
+ err = storage.StoreTrie(ts)
+ require.NoError(t, err)
+
+ // Clear trie from cache and fetch data from disk.
+ storage.lock.Lock()
+ delete(storage.tries, root)
+ storage.lock.Unlock()
+
+ data, err := storage.GetStorage(&root, trieKV[0].key)
+ require.NoError(t, err)
+ require.Equal(t, trieKV[0].value, data)
+
+ storage.lock.Lock()
+ delete(storage.tries, root)
+ storage.lock.Unlock()
+
+ prefixKeys, err := storage.GetKeysWithPrefix(&root, []byte("ke"))
+ require.NoError(t, err)
+ require.Equal(t, 2, len(prefixKeys))
+
+ storage.lock.Lock()
+ delete(storage.tries, root)
+ storage.lock.Unlock()
+
+ entries, err := storage.Entries(&root)
+ require.NoError(t, err)
+ require.Equal(t, 3, len(entries))
+}
diff --git a/lib/trie/child_storage.go b/lib/trie/child_storage.go
index 85d8b4a1e6..4af75a532c 100644
--- a/lib/trie/child_storage.go
+++ b/lib/trie/child_storage.go
@@ -36,7 +36,7 @@ func (t *Trie) PutChild(keyToChild []byte, child *Trie) error {
value := [32]byte(childHash)
t.Put(key, value[:])
- t.children[childHash] = child
+ t.childTries[childHash] = child
return nil
}
@@ -50,7 +50,7 @@ func (t *Trie) GetChild(keyToChild []byte) (*Trie, error) {
hash := [32]byte{}
copy(hash[:], childHash)
- return t.children[common.Hash(hash)], nil
+ return t.childTries[common.Hash(hash)], nil
}
// PutIntoChild puts a key-value pair into the child trie located in the main trie at key :child_storage:[keyToChild]
@@ -71,8 +71,8 @@ func (t *Trie) PutIntoChild(keyToChild, key, value []byte) error {
return err
}
- t.children[origChildHash] = nil
- t.children[childHash] = child
+ t.childTries[origChildHash] = nil
+ t.childTries[childHash] = child
return t.PutChild(keyToChild, child)
}
diff --git a/lib/trie/trie.go b/lib/trie/trie.go
index 48dd9e5a09..d1db892612 100644
--- a/lib/trie/trie.go
+++ b/lib/trie/trie.go
@@ -31,7 +31,7 @@ var EmptyHash, _ = NewEmptyTrie().Hash()
type Trie struct {
generation uint64
root node
- children map[common.Hash]*Trie // Used to store the child tries.
+ childTries map[common.Hash]*Trie // Used to store the child tries.
}
// NewEmptyTrie creates a trie with a nil root
@@ -43,7 +43,7 @@ func NewEmptyTrie() *Trie {
func NewTrie(root node) *Trie {
return &Trie{
root: root,
- children: make(map[common.Hash]*Trie),
+ childTries: make(map[common.Hash]*Trie),
generation: 0, // Initially zero but increases after every snapshot.
}
}
@@ -53,7 +53,7 @@ func (t *Trie) Snapshot() *Trie {
oldTrie := &Trie{
generation: t.generation,
root: t.root,
- children: t.children,
+ childTries: t.childTries,
}
t.generation++
return oldTrie
From 3880a40839a32624e44bf4f8b7b990266b0a6900 Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Thu, 25 Mar 2021 15:04:23 -0400
Subject: [PATCH 04/21] fix(dot/state): fix usage of trie.Snapshot (#1489)
---
dot/state/storage.go | 21 ++++++++++++++++-----
dot/state/storage_test.go | 23 +++++++++++++++++++++++
dot/sync/syncer.go | 1 -
lib/babe/babe.go | 1 -
lib/runtime/storage/trie.go | 7 ++++---
5 files changed, 43 insertions(+), 10 deletions(-)
diff --git a/dot/state/storage.go b/dot/state/storage.go
index 5c9a45be54..30659447c1 100644
--- a/dot/state/storage.go
+++ b/dot/state/storage.go
@@ -43,7 +43,7 @@ func errTrieDoesNotExist(hash common.Hash) error {
// StorageState is the struct that holds the trie, db and lock
type StorageState struct {
blockState *BlockState
- tries map[common.Hash]*trie.Trie
+ tries map[common.Hash]*trie.Trie // map of root -> trie
baseDB chaindb.Database
db chaindb.Database
@@ -126,16 +126,27 @@ func (s *StorageState) TrieState(root *common.Hash) (*rtstorage.TrieState, error
t := s.tries[*root]
s.lock.RUnlock()
- if t != nil && t.MustHash() == *root { // TODO: figure out why it seems like snapshotted tries are getting modified
- return rtstorage.NewTrieState(t)
+ if t != nil && t.MustHash() != *root {
+ panic("trie does not have expected root")
}
- tr, err := s.LoadFromDB(*root)
+ if t == nil {
+ var err error
+ t, err = s.LoadFromDB(*root)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ curr, err := rtstorage.NewTrieState(t)
if err != nil {
return nil, err
}
- return rtstorage.NewTrieState(tr)
+ s.lock.Lock()
+ s.tries[*root] = curr.Snapshot()
+ s.lock.Unlock()
+ return curr, nil
}
// StoreInDB encodes the entire trie and writes it to the DB
diff --git a/dot/state/storage_test.go b/dot/state/storage_test.go
index 5c55c4e5c8..5be980bea2 100644
--- a/dot/state/storage_test.go
+++ b/dot/state/storage_test.go
@@ -37,6 +37,7 @@ func TestStorage_StoreAndLoadTrie(t *testing.T) {
require.NoError(t, err)
ts2, err := runtime.NewTrieState(trie)
require.NoError(t, err)
+ ts2.Snapshot()
require.Equal(t, ts.Trie(), ts2.Trie())
}
@@ -70,6 +71,28 @@ func TestStorage_GetStorageByBlockHash(t *testing.T) {
require.Equal(t, value, res)
}
+func TestStorage_TrieState(t *testing.T) {
+ storage := newTestStorageState(t)
+ ts, err := storage.TrieState(&trie.EmptyHash)
+ require.NoError(t, err)
+ ts.Set([]byte("noot"), []byte("washere"))
+
+ root, err := ts.Root()
+ require.NoError(t, err)
+ err = storage.StoreTrie(ts)
+ require.NoError(t, err)
+
+ time.Sleep(time.Millisecond * 100)
+
+ // get trie from db
+ storage.lock.Lock()
+ delete(storage.tries, root)
+ storage.lock.Unlock()
+ ts3, err := storage.TrieState(&root)
+ require.NoError(t, err)
+ require.Equal(t, ts.Trie().MustHash(), ts3.Trie().MustHash())
+}
+
func TestStorage_LoadFromDB(t *testing.T) {
storage := newTestStorageState(t)
ts, err := storage.TrieState(&trie.EmptyHash)
diff --git a/dot/sync/syncer.go b/dot/sync/syncer.go
index fb92a91b37..ce0df185a3 100644
--- a/dot/sync/syncer.go
+++ b/dot/sync/syncer.go
@@ -303,7 +303,6 @@ func (s *Service) handleBlock(block *types.Block) error {
return err
}
- ts.Snapshot()
root := ts.MustRoot()
if !bytes.Equal(parent.StateRoot[:], root[:]) {
panic("parent state root does not match snapshot state root")
diff --git a/lib/babe/babe.go b/lib/babe/babe.go
index 47da02b7c1..fb937b88cd 100644
--- a/lib/babe/babe.go
+++ b/lib/babe/babe.go
@@ -463,7 +463,6 @@ func (b *Service) handleSlot(slotNum uint64) error {
return err
}
- ts.Snapshot()
b.rt.SetContextStorage(ts)
block, err := b.buildBlock(parent, currentSlot)
diff --git a/lib/runtime/storage/trie.go b/lib/runtime/storage/trie.go
index 6663f9041b..4ea5a018a7 100644
--- a/lib/runtime/storage/trie.go
+++ b/lib/runtime/storage/trie.go
@@ -52,8 +52,9 @@ func (s *TrieState) Trie() *trie.Trie {
// Snapshot creates a new "version" of the trie. The trie before Snapshot is called
// can no longer be modified, all further changes are on a new "version" of the trie.
-func (s *TrieState) Snapshot() {
- _ = s.t.Snapshot()
+// It returns the previous version of the trie.
+func (s *TrieState) Snapshot() *trie.Trie {
+ return s.t.Snapshot()
}
// BeginStorageTransaction begins a new nested storage transaction which will either be committed or rolled back at a later time.
@@ -79,7 +80,7 @@ func (s *TrieState) RollbackStorageTransaction() {
}
// Set sets a key-value pair in the trie
-func (s *TrieState) Set(key []byte, value []byte) {
+func (s *TrieState) Set(key, value []byte) {
s.lock.Lock()
defer s.lock.Unlock()
s.t.Put(key, value)
From 3b2ad8d0b7eb9159d862fa8188423914da40f29e Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Thu, 25 Mar 2021 16:05:15 -0400
Subject: [PATCH 05/21] fix(dot/state,dot/network): improve memory usage when
syncing (#1491)
---
dot/core/test_helpers.go | 2 ++
dot/network/service_test.go | 2 +-
dot/network/state.go | 2 ++
dot/network/sync.go | 17 ++++++++++++-----
dot/network/test_helpers.go | 4 ++--
dot/node.go | 8 ++++++++
dot/rpc/modules/system_test.go | 2 ++
dot/state/storage.go | 14 ++++++++++++++
dot/state/storage_test.go | 30 ++++++++++++++++++++++++++++++
dot/sync/interface.go | 1 +
dot/sync/syncer.go | 6 ++++++
tests/rpc/rpc_01-system_test.go | 2 +-
12 files changed, 81 insertions(+), 9 deletions(-)
diff --git a/dot/core/test_helpers.go b/dot/core/test_helpers.go
index 3da5758d64..e5657c6a28 100644
--- a/dot/core/test_helpers.go
+++ b/dot/core/test_helpers.go
@@ -261,6 +261,8 @@ func (s *mockSyncer) IsSynced() bool {
return false
}
+func (s *mockSyncer) SetSyncing(bool) {}
+
type mockDigestItem struct { //nolint
i int
}
diff --git a/dot/network/service_test.go b/dot/network/service_test.go
index 54f4c7b413..dab4ccb26c 100644
--- a/dot/network/service_test.go
+++ b/dot/network/service_test.go
@@ -187,7 +187,7 @@ func TestService_Health(t *testing.T) {
require.Equal(t, s.Health().IsSyncing, true)
mockSync := s.syncer.(*mockSyncer)
- mockSync.setSyncedState(true)
+ mockSync.SetSyncing(false)
require.Equal(t, s.Health().IsSyncing, false)
}
diff --git a/dot/network/state.go b/dot/network/state.go
index 2077ada3b3..77725bd7de 100644
--- a/dot/network/state.go
+++ b/dot/network/state.go
@@ -46,6 +46,8 @@ type Syncer interface {
// IsSynced exposes the internal synced state // TODO: use syncQueue for this
IsSynced() bool
+
+ SetSyncing(bool)
}
// TransactionHandler is the interface used by the transactions sub-protocol
diff --git a/dot/network/sync.go b/dot/network/sync.go
index af0b02c3ab..a456e22df8 100644
--- a/dot/network/sync.go
+++ b/dot/network/sync.go
@@ -164,6 +164,8 @@ func (q *syncQueue) syncAtHead() {
return
}
+ q.s.syncer.SetSyncing(true)
+
for {
select {
// sleep for average block time TODO: make this configurable from slot duration
@@ -183,6 +185,9 @@ func (q *syncQueue) syncAtHead() {
continue
}
+ logger.Info("set syncing to false")
+ q.s.syncer.SetSyncing(false)
+
// we have received new blocks since the last check, sleep
if prev.Number.Int64() < curr.Number.Int64() {
prev = curr
@@ -192,10 +197,7 @@ func (q *syncQueue) syncAtHead() {
prev = curr
start := uint64(curr.Number.Int64()) + 1
logger.Debug("haven't received new blocks since last check, pushing request", "start", start)
- q.requestData.Store(start, requestData{
- sent: true,
- received: false,
- })
+ q.requestData.Delete(start)
q.pushRequest(start, 1, "")
}
}
@@ -248,7 +250,12 @@ func (q *syncQueue) handleResponseQueue() {
// prune peers with low score and connect to new peers
func (q *syncQueue) prunePeers() {
for {
- time.Sleep(time.Second * 30)
+ select {
+ case <-time.After(time.Second * 30):
+ case <-q.ctx.Done():
+ return
+ }
+
logger.Debug("✂️ pruning peers w/ low score...")
peers := q.getSortedPeers()
diff --git a/dot/network/test_helpers.go b/dot/network/test_helpers.go
index 17df3c7a10..03851ac325 100644
--- a/dot/network/test_helpers.go
+++ b/dot/network/test_helpers.go
@@ -65,8 +65,8 @@ func (s *mockSyncer) IsSynced() bool {
return s.synced
}
-func (s *mockSyncer) setSyncedState(newState bool) {
- s.synced = newState
+func (s *mockSyncer) SetSyncing(syncing bool) {
+ s.synced = !syncing
}
type testStreamHandler struct {
diff --git a/dot/node.go b/dot/node.go
index 53ef7103c4..34d0abcdcb 100644
--- a/dot/node.go
+++ b/dot/node.go
@@ -22,6 +22,7 @@ import (
"os"
"os/signal"
"path"
+ "runtime/debug"
"sync"
"syscall"
@@ -158,6 +159,13 @@ func NodeInitialized(basepath string, expected bool) bool {
// NewNode creates a new dot node from a dot node configuration
func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node, error) {
+ // set garbage collection percent to 10%
+ // can be overwritten by setting the GOGC env veriable, which defaults to 100
+ prev := debug.SetGCPercent(10)
+ if prev != 100 {
+ debug.SetGCPercent(prev)
+ }
+
setupLogger(cfg)
// if authority node, should have at least 1 key in keystore
diff --git a/dot/rpc/modules/system_test.go b/dot/rpc/modules/system_test.go
index a9d6c207af..08e3512faf 100644
--- a/dot/rpc/modules/system_test.go
+++ b/dot/rpc/modules/system_test.go
@@ -63,6 +63,8 @@ func (s *mockSyncer) IsSynced() bool {
return false
}
+func (s *mockSyncer) SetSyncing(_ bool) {}
+
type mockBlockState struct{}
func (s *mockBlockState) BestBlockHeader() (*types.Header, error) {
diff --git a/dot/state/storage.go b/dot/state/storage.go
index 30659447c1..73bea66666 100644
--- a/dot/state/storage.go
+++ b/dot/state/storage.go
@@ -52,6 +52,8 @@ type StorageState struct {
// change notifiers
changedLock sync.RWMutex
subscriptions map[byte]*StorageSubscription
+
+ syncing bool
}
// NewStorageState creates a new StorageState backed by the given trie and database located at basePath.
@@ -76,6 +78,11 @@ func NewStorageState(db chaindb.Database, blockState *BlockState, t *trie.Trie)
}, nil
}
+// SetSyncing sets whether the node is currently syncing or not
+func (s *StorageState) SetSyncing(syncing bool) {
+ s.syncing = syncing
+}
+
func (s *StorageState) pruneKey(keyHeader *types.Header) {
s.lock.Lock()
defer s.lock.Unlock()
@@ -93,6 +100,12 @@ func (s *StorageState) pruneKey(keyHeader *types.Header) {
func (s *StorageState) StoreTrie(ts *rtstorage.TrieState) error {
s.lock.Lock()
root := ts.MustRoot()
+ if s.syncing {
+ // keep only the trie at the head of the chain when syncing
+ for key := range s.tries {
+ delete(s.tries, key)
+ }
+ }
s.tries[root] = ts.Trie()
s.lock.Unlock()
@@ -108,6 +121,7 @@ func (s *StorageState) StoreTrie(ts *rtstorage.TrieState) error {
logger.Warn("failed to notify storage subscriptions", "error", err)
}
}()
+
return nil
}
diff --git a/dot/state/storage_test.go b/dot/state/storage_test.go
index 5be980bea2..f6601eebdf 100644
--- a/dot/state/storage_test.go
+++ b/dot/state/storage_test.go
@@ -143,3 +143,33 @@ func TestStorage_LoadFromDB(t *testing.T) {
require.NoError(t, err)
require.Equal(t, 3, len(entries))
}
+
+func TestStorage_StoreTrie_Syncing(t *testing.T) {
+ storage := newTestStorageState(t)
+ ts, err := storage.TrieState(&trie.EmptyHash)
+ require.NoError(t, err)
+
+ key := []byte("testkey")
+ value := []byte("testvalue")
+ ts.Set(key, value)
+
+ storage.SetSyncing(true)
+ err = storage.StoreTrie(ts)
+ require.NoError(t, err)
+ require.Equal(t, 1, len(storage.tries))
+}
+
+func TestStorage_StoreTrie_NotSyncing(t *testing.T) {
+ storage := newTestStorageState(t)
+ ts, err := storage.TrieState(&trie.EmptyHash)
+ require.NoError(t, err)
+
+ key := []byte("testkey")
+ value := []byte("testvalue")
+ ts.Set(key, value)
+
+ storage.SetSyncing(false)
+ err = storage.StoreTrie(ts)
+ require.NoError(t, err)
+ require.Equal(t, 2, len(storage.tries))
+}
diff --git a/dot/sync/interface.go b/dot/sync/interface.go
index 2500d0df5d..1a457091e2 100644
--- a/dot/sync/interface.go
+++ b/dot/sync/interface.go
@@ -52,6 +52,7 @@ type StorageState interface {
TrieState(root *common.Hash) (*rtstorage.TrieState, error)
StoreTrie(ts *rtstorage.TrieState) error
LoadCodeHash(*common.Hash) (common.Hash, error)
+ SetSyncing(bool)
}
// TransactionState is the interface for transaction queue methods
diff --git a/dot/sync/syncer.go b/dot/sync/syncer.go
index ce0df185a3..c5a96dd23c 100644
--- a/dot/sync/syncer.go
+++ b/dot/sync/syncer.go
@@ -411,3 +411,9 @@ func (s *Service) handleDigests(header *types.Header) {
func (s *Service) IsSynced() bool {
return s.synced
}
+
+// SetSyncing sets whether the node is currently syncing or not
+func (s *Service) SetSyncing(syncing bool) {
+ s.synced = !syncing
+ s.storageState.SetSyncing(syncing)
+}
diff --git a/tests/rpc/rpc_01-system_test.go b/tests/rpc/rpc_01-system_test.go
index 48c4d2ab1d..b64641b1a0 100644
--- a/tests/rpc/rpc_01-system_test.go
+++ b/tests/rpc/rpc_01-system_test.go
@@ -60,7 +60,7 @@ func TestSystemRPC(t *testing.T) {
expected: modules.SystemHealthResponse{
Peers: 2,
- IsSyncing: false,
+ IsSyncing: true,
ShouldHavePeers: true,
},
params: "{}",
From 22f72692dac1384bb2b6581b90fe905b2462760f Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Thu, 25 Mar 2021 18:53:18 -0400
Subject: [PATCH 06/21] fix(dot/network): update notificationsProtocol
handshakeData to sync.Map (#1492)
---
dot/network/block_announce.go | 8 +++----
dot/network/block_announce_test.go | 5 ++--
dot/network/host_test.go | 8 +++----
dot/network/notifications.go | 38 +++++++++++++++++++-----------
dot/network/notifications_test.go | 30 +++++++++++++----------
dot/network/service.go | 27 ++++++++++-----------
dot/network/service_test.go | 4 ++--
7 files changed, 67 insertions(+), 53 deletions(-)
diff --git a/dot/network/block_announce.go b/dot/network/block_announce.go
index 541e7dbf3a..ab31795702 100644
--- a/dot/network/block_announce.go
+++ b/dot/network/block_announce.go
@@ -228,13 +228,13 @@ func (s *Service) validateBlockAnnounceHandshake(peer peer.ID, hs Handshake) err
// don't need to lock here, since function is always called inside the func returned by
// `createNotificationsMessageHandler` which locks the map beforehand.
- data, ok := np.handshakeData[peer]
+ data, ok := np.getHandshakeData(peer)
if !ok {
- np.handshakeData[peer] = &handshakeData{
+ np.handshakeData.Store(peer, &handshakeData{
received: true,
validated: true,
- }
- data = np.handshakeData[peer]
+ })
+ data, _ = np.getHandshakeData(peer)
}
data.handshake = hs
diff --git a/dot/network/block_announce_test.go b/dot/network/block_announce_test.go
index 8e01dca514..ea381dad6d 100644
--- a/dot/network/block_announce_test.go
+++ b/dot/network/block_announce_test.go
@@ -18,6 +18,7 @@ package network
import (
"math/big"
+ "sync"
"testing"
"github.com/ChainSafe/gossamer/dot/types"
@@ -116,10 +117,10 @@ func TestValidateBlockAnnounceHandshake(t *testing.T) {
nodeA := createTestService(t, configA)
nodeA.noGossip = true
nodeA.notificationsProtocols[BlockAnnounceMsgType] = ¬ificationsProtocol{
- handshakeData: make(map[peer.ID]*handshakeData),
+ handshakeData: new(sync.Map),
}
testPeerID := peer.ID("noot")
- nodeA.notificationsProtocols[BlockAnnounceMsgType].handshakeData[testPeerID] = &handshakeData{}
+ nodeA.notificationsProtocols[BlockAnnounceMsgType].handshakeData.Store(testPeerID, &handshakeData{})
err := nodeA.validateBlockAnnounceHandshake(testPeerID, &BlockAnnounceHandshake{
BestBlockNumber: 100,
diff --git a/dot/network/host_test.go b/dot/network/host_test.go
index 0a9f010b57..56430d94cb 100644
--- a/dot/network/host_test.go
+++ b/dot/network/host_test.go
@@ -363,13 +363,13 @@ func TestStreamCloseMetadataCleanup(t *testing.T) {
info := nodeA.notificationsProtocols[BlockAnnounceMsgType]
// Set handshake data to received
- info.handshakeData[nodeB.host.id()] = &handshakeData{
+ info.handshakeData.Store(nodeB.host.id(), &handshakeData{
received: true,
validated: true,
- }
+ })
// Verify that handshake data exists.
- _, ok := info.handshakeData[nodeB.host.id()]
+ _, ok := info.getHandshakeData(nodeB.host.id())
require.True(t, ok)
time.Sleep(time.Second)
@@ -379,7 +379,7 @@ func TestStreamCloseMetadataCleanup(t *testing.T) {
time.Sleep(time.Second)
// Verify that handshake data is cleared.
- _, ok = info.handshakeData[nodeB.host.id()]
+ _, ok = info.getHandshakeData(nodeB.host.id())
require.False(t, ok)
}
diff --git a/dot/network/notifications.go b/dot/network/notifications.go
index bfdff68df9..a733d06b78 100644
--- a/dot/network/notifications.go
+++ b/dot/network/notifications.go
@@ -54,10 +54,19 @@ type (
type notificationsProtocol struct {
protocolID protocol.ID
getHandshake HandshakeGetter
- handshakeData map[peer.ID]*handshakeData
+ handshakeData *sync.Map //map[peer.ID]*handshakeData
mapMu sync.RWMutex
}
+func (n *notificationsProtocol) getHandshakeData(pid peer.ID) (*handshakeData, bool) {
+ data, has := n.handshakeData.Load(pid)
+ if !has {
+ return nil, false
+ }
+
+ return data.(*handshakeData), true
+}
+
type handshakeData struct {
received bool
validated bool
@@ -72,7 +81,7 @@ func createDecoder(info *notificationsProtocol, handshakeDecoder HandshakeDecode
info.mapMu.RLock()
defer info.mapMu.RUnlock()
- if hsData, has := info.handshakeData[peer]; !has || !hsData.received {
+ if hsData, has := info.getHandshakeData(peer); !has || !hsData.received {
return handshakeDecoder(in)
}
@@ -112,12 +121,12 @@ func (s *Service) createNotificationsMessageHandler(info *notificationsProtocol,
defer info.mapMu.Unlock()
// if we are the receiver and haven't received the handshake already, validate it
- if _, has := info.handshakeData[peer]; !has {
+ if _, has := info.getHandshakeData(peer); !has {
logger.Trace("receiver: validating handshake", "protocol", info.protocolID)
- info.handshakeData[peer] = &handshakeData{
+ info.handshakeData.Store(peer, &handshakeData{
validated: false,
received: true,
- }
+ })
err := handshakeValidator(peer, hs)
if err != nil {
@@ -126,7 +135,8 @@ func (s *Service) createNotificationsMessageHandler(info *notificationsProtocol,
return errCannotValidateHandshake
}
- info.handshakeData[peer].validated = true
+ data, _ := info.getHandshakeData(peer)
+ data.validated = true
// once validated, send back a handshake
resp, err := info.getHandshake()
@@ -145,25 +155,25 @@ func (s *Service) createNotificationsMessageHandler(info *notificationsProtocol,
}
// if we are the initiator and haven't received the handshake already, validate it
- if hsData, has := info.handshakeData[peer]; has && !hsData.validated {
+ if hsData, has := info.getHandshakeData(peer); has && !hsData.validated {
logger.Trace("sender: validating handshake")
err := handshakeValidator(peer, hs)
if err != nil {
logger.Trace("failed to validate handshake", "protocol", info.protocolID, "peer", peer, "error", err)
- info.handshakeData[peer].validated = false
+ hsData.validated = false
_ = stream.Conn().Close()
return errCannotValidateHandshake
}
- info.handshakeData[peer].validated = true
- info.handshakeData[peer].received = true
+ hsData.validated = true
+ hsData.received = true
logger.Trace("sender: validated handshake", "protocol", info.protocolID, "peer", peer)
} else if hsData.received {
return nil
}
// if we are the initiator, send the message
- if hsData, has := info.handshakeData[peer]; has && hsData.validated && hsData.received && hsData.outboundMsg != nil {
+ if hsData, has := info.getHandshakeData(peer); has && hsData.validated && hsData.received && hsData.outboundMsg != nil {
logger.Trace("sender: sending message", "protocol", info.protocolID)
err := s.host.send(peer, info.protocolID, hsData.outboundMsg)
if err != nil {
@@ -223,11 +233,11 @@ func (s *Service) broadcastExcluding(info *notificationsProtocol, excluding peer
info.mapMu.RLock()
defer info.mapMu.RUnlock()
- if hsData, has := info.handshakeData[peer]; !has || !hsData.received {
- info.handshakeData[peer] = &handshakeData{
+ if hsData, has := info.getHandshakeData(peer); !has || !hsData.received {
+ info.handshakeData.Store(peer, &handshakeData{
validated: false,
outboundMsg: msg,
- }
+ })
logger.Trace("sending handshake", "protocol", info.protocolID, "peer", peer, "message", hs)
err = s.host.send(peer, info.protocolID, hs)
diff --git a/dot/network/notifications_test.go b/dot/network/notifications_test.go
index 2eeeadcb6f..f0f65f793e 100644
--- a/dot/network/notifications_test.go
+++ b/dot/network/notifications_test.go
@@ -18,6 +18,7 @@ package network
import (
"math/big"
+ "sync"
"testing"
"time"
@@ -46,15 +47,15 @@ func TestCreateDecoder_BlockAnnounce(t *testing.T) {
info := ¬ificationsProtocol{
protocolID: s.host.protocolID + blockAnnounceID,
getHandshake: s.getBlockAnnounceHandshake,
- handshakeData: make(map[peer.ID]*handshakeData),
+ handshakeData: new(sync.Map),
}
decoder := createDecoder(info, decodeBlockAnnounceHandshake, decodeBlockAnnounceMessage)
// haven't received handshake from peer
testPeerID := peer.ID("QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ")
- info.handshakeData[testPeerID] = &handshakeData{
+ info.handshakeData.Store(testPeerID, &handshakeData{
received: false,
- }
+ })
testHandshake := &BlockAnnounceHandshake{
Roles: 4,
@@ -82,7 +83,8 @@ func TestCreateDecoder_BlockAnnounce(t *testing.T) {
require.NoError(t, err)
// set handshake data to received
- info.handshakeData[testPeerID].received = true
+ hsData, _ := info.getHandshakeData(testPeerID)
+ hsData.received = true
msg, err = decoder(enc, testPeerID)
require.NoError(t, err)
require.Equal(t, testBlockAnnounce, msg)
@@ -132,15 +134,15 @@ func TestCreateNotificationsMessageHandler_BlockAnnounce(t *testing.T) {
info := ¬ificationsProtocol{
protocolID: s.host.protocolID + blockAnnounceID,
getHandshake: s.getBlockAnnounceHandshake,
- handshakeData: make(map[peer.ID]*handshakeData),
+ handshakeData: new(sync.Map),
}
handler := s.createNotificationsMessageHandler(info, s.validateBlockAnnounceHandshake, s.handleBlockAnnounceMessage)
// set handshake data to received
- info.handshakeData[testPeerID] = &handshakeData{
+ info.handshakeData.Store(testPeerID, &handshakeData{
received: true,
validated: true,
- }
+ })
msg := &BlockAnnounceMessage{
Number: big.NewInt(10),
}
@@ -164,7 +166,7 @@ func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T)
info := ¬ificationsProtocol{
protocolID: s.host.protocolID + blockAnnounceID,
getHandshake: s.getBlockAnnounceHandshake,
- handshakeData: make(map[peer.ID]*handshakeData),
+ handshakeData: new(sync.Map),
}
handler := s.createNotificationsMessageHandler(info, s.validateBlockAnnounceHandshake, s.handleBlockAnnounceMessage)
@@ -205,8 +207,10 @@ func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T)
err = handler(stream, testHandshake)
require.Equal(t, errCannotValidateHandshake, err)
- require.True(t, info.handshakeData[testPeerID].received)
- require.False(t, info.handshakeData[testPeerID].validated)
+ data, has := info.getHandshakeData(testPeerID)
+ require.True(t, has)
+ require.True(t, data.received)
+ require.False(t, data.validated)
// try valid handshake
testHandshake = &BlockAnnounceHandshake{
@@ -218,6 +222,8 @@ func TestCreateNotificationsMessageHandler_BlockAnnounceHandshake(t *testing.T)
err = handler(stream, testHandshake)
require.NoError(t, err)
- require.True(t, info.handshakeData[testPeerID].received)
- require.True(t, info.handshakeData[testPeerID].validated)
+ data, has = info.getHandshakeData(testPeerID)
+ require.True(t, has)
+ require.True(t, data.received)
+ require.True(t, data.validated)
}
diff --git a/dot/network/service.go b/dot/network/service.go
index 6ce82d8be0..a970812985 100644
--- a/dot/network/service.go
+++ b/dot/network/service.go
@@ -304,10 +304,10 @@ func (s *Service) handleConn(conn libp2pnetwork.Conn) {
defer info.mapMu.RUnlock()
peer := conn.RemotePeer()
- if hsData, has := info.handshakeData[peer]; !has || !hsData.received {
- info.handshakeData[peer] = &handshakeData{
+ if hsData, has := info.getHandshakeData(peer); !has || !hsData.received {
+ info.handshakeData.Store(peer, &handshakeData{
validated: false,
- }
+ })
logger.Trace("sending handshake", "protocol", info.protocolID, "peer", peer, "message", hs)
err = s.host.send(peer, info.protocolID, hs)
@@ -407,7 +407,7 @@ func (s *Service) RegisterNotificationsProtocol(sub protocol.ID,
np := ¬ificationsProtocol{
protocolID: protocolID,
getHandshake: handshakeGetter,
- handshakeData: make(map[peer.ID]*handshakeData),
+ handshakeData: new(sync.Map),
}
s.notificationsProtocols[messageID] = np
@@ -416,13 +416,13 @@ func (s *Service) RegisterNotificationsProtocol(sub protocol.ID,
np.mapMu.Lock()
defer np.mapMu.Unlock()
- if _, ok := np.handshakeData[peerID]; ok {
+ if _, ok := np.getHandshakeData(peerID); ok {
logger.Trace(
"Cleaning up handshake data",
"peer", peerID,
"protocol", protocolID,
)
- delete(np.handshakeData, peerID)
+ np.handshakeData.Delete(peerID)
}
})
@@ -625,24 +625,20 @@ func (s *Service) Peers() []common.PeerInfo {
peers := []common.PeerInfo{}
s.notificationsMu.RLock()
- defer s.notificationsMu.RUnlock()
+ np := s.notificationsProtocols[BlockAnnounceMsgType]
+ s.notificationsMu.RUnlock()
for _, p := range s.host.peers() {
- if s.notificationsProtocols[BlockAnnounceMsgType].handshakeData[p] == nil {
+ data, has := np.getHandshakeData(p)
+ if !has || data.handshake == nil {
peers = append(peers, common.PeerInfo{
PeerID: p.String(),
})
continue
}
- peerHandshakeMessage := s.notificationsProtocols[BlockAnnounceMsgType].handshakeData[p].handshake
- if peerHandshakeMessage == nil {
- peers = append(peers, common.PeerInfo{
- PeerID: p.String(),
- })
- continue
- }
+ peerHandshakeMessage := data.handshake
peers = append(peers, common.PeerInfo{
PeerID: p.String(),
Roles: peerHandshakeMessage.(*BlockAnnounceHandshake).Roles,
@@ -650,6 +646,7 @@ func (s *Service) Peers() []common.PeerInfo {
BestNumber: uint64(peerHandshakeMessage.(*BlockAnnounceHandshake).BestBlockNumber),
})
}
+
return peers
}
diff --git a/dot/network/service_test.go b/dot/network/service_test.go
index dab4ccb26c..4f9743fc1f 100644
--- a/dot/network/service_test.go
+++ b/dot/network/service_test.go
@@ -371,13 +371,13 @@ func TestHandleConn(t *testing.T) {
require.Equal(t, 1, aScore)
infoA := nodeA.notificationsProtocols[BlockAnnounceMsgType]
- hsDataB, has := infoA.handshakeData[nodeB.host.id()]
+ hsDataB, has := infoA.getHandshakeData(nodeB.host.id())
require.True(t, has)
require.True(t, hsDataB.received)
require.True(t, hsDataB.validated)
infoB := nodeB.notificationsProtocols[BlockAnnounceMsgType]
- hsDataA, has := infoB.handshakeData[nodeA.host.id()]
+ hsDataA, has := infoB.getHandshakeData(nodeA.host.id())
require.True(t, has)
require.True(t, hsDataA.received)
require.True(t, hsDataA.validated)
From a6e0bd07a2cbb6015f83764f284dc45f5922e4d3 Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Fri, 26 Mar 2021 09:49:20 -0400
Subject: [PATCH 07/21] remove log (#1493)
---
dot/network/sync.go | 1 -
1 file changed, 1 deletion(-)
diff --git a/dot/network/sync.go b/dot/network/sync.go
index a456e22df8..70d822d577 100644
--- a/dot/network/sync.go
+++ b/dot/network/sync.go
@@ -185,7 +185,6 @@ func (q *syncQueue) syncAtHead() {
continue
}
- logger.Info("set syncing to false")
q.s.syncer.SetSyncing(false)
// we have received new blocks since the last check, sleep
From ae7012b2cb4c39ccd4a5cc2220795929c853ae00 Mon Sep 17 00:00:00 2001
From: noot <36753753+noot@users.noreply.github.com>
Date: Mon, 5 Apr 2021 11:38:43 -0400
Subject: [PATCH 08/21] feat(dot/network): request block justifications when
near head (#1499)
---
dot/network/block_announce.go | 16 +--
dot/network/host.go | 8 +-
dot/network/host_test.go | 10 +-
dot/network/notifications.go | 15 ++-
dot/network/service.go | 11 +-
dot/network/state.go | 1 +
dot/network/state_test.go | 4 +
dot/network/sync.go | 144 +++++++++++++++------
dot/network/sync_justification.go | 89 +++++++++++++
dot/network/sync_justification_test.go | 168 +++++++++++++++++++++++++
dot/network/transaction.go | 26 +---
dot/network/transaction_test.go | 4 +-
dot/network/utils.go | 4 +-
dot/rpc/modules/system_test.go | 4 +
dot/state/block.go | 10 ++
dot/state/block_test.go | 26 ++++
dot/sync/syncer.go | 7 +-
lib/grandpa/network.go | 2 +-
18 files changed, 455 insertions(+), 94 deletions(-)
create mode 100644 dot/network/sync_justification.go
create mode 100644 dot/network/sync_justification_test.go
diff --git a/dot/network/block_announce.go b/dot/network/block_announce.go
index ab31795702..05e1c9b71f 100644
--- a/dot/network/block_announce.go
+++ b/dot/network/block_announce.go
@@ -212,14 +212,6 @@ func (s *Service) validateBlockAnnounceHandshake(peer peer.ID, hs Handshake) err
return errors.New("genesis hash mismatch")
}
- // if peer has higher best block than us, begin syncing
- latestHeader, err := s.blockState.BestBlockHeader()
- if err != nil {
- return err
- }
-
- bestBlockNum := big.NewInt(int64(bhs.BestBlockNumber))
-
np, ok := s.notificationsProtocols[BlockAnnounceMsgType]
if !ok {
// this should never happen.
@@ -239,6 +231,14 @@ func (s *Service) validateBlockAnnounceHandshake(peer peer.ID, hs Handshake) err
data.handshake = hs
+ // if peer has higher best block than us, begin syncing
+ latestHeader, err := s.blockState.BestBlockHeader()
+ if err != nil {
+ return err
+ }
+
+ bestBlockNum := big.NewInt(int64(bhs.BestBlockNumber))
+
// check if peer block number is greater than host block number
if latestHeader.Number.Cmp(bestBlockNum) >= 0 {
return nil
diff --git a/dot/network/host.go b/dot/network/host.go
index 787b0c0e96..79594760e4 100644
--- a/dot/network/host.go
+++ b/dot/network/host.go
@@ -237,7 +237,7 @@ func (h *host) bootstrap() {
// peer (gets the already opened outbound message stream or opens a new one).
func (h *host) send(p peer.ID, pid protocol.ID, msg Message) (err error) {
// get outbound stream for given peer
- s := h.getStream(p, pid)
+ s := h.getOutboundStream(p, pid)
// check if stream needs to be opened
if s == nil {
@@ -286,10 +286,10 @@ func (h *host) writeToStream(s libp2pnetwork.Stream, msg Message) error {
return err
}
-// getStream returns the outbound message stream for the given peer or returns
+// getOutboundStream returns the outbound message stream for the given peer or returns
// nil if no outbound message stream exists. For each peer, each host opens an
// outbound message stream and writes to the same stream until closed or reset.
-func (h *host) getStream(p peer.ID, pid protocol.ID) (stream libp2pnetwork.Stream) {
+func (h *host) getOutboundStream(p peer.ID, pid protocol.ID) (stream libp2pnetwork.Stream) {
conns := h.h.Network().ConnsToPeer(p)
// loop through connections (only one for now)
@@ -310,7 +310,7 @@ func (h *host) getStream(p peer.ID, pid protocol.ID) (stream libp2pnetwork.Strea
// closeStream closes a stream open to the peer with the given sub-protocol, if it exists.
func (h *host) closeStream(p peer.ID, pid protocol.ID) {
- stream := h.getStream(p, pid)
+ stream := h.getOutboundStream(p, pid)
if stream != nil {
_ = stream.Close()
}
diff --git a/dot/network/host_test.go b/dot/network/host_test.go
index 56430d94cb..aca5f2313e 100644
--- a/dot/network/host_test.go
+++ b/dot/network/host_test.go
@@ -269,7 +269,7 @@ func TestExistingStream(t *testing.T) {
}
require.NoError(t, err)
- stream := nodeA.host.getStream(nodeB.host.id(), nodeB.host.protocolID)
+ stream := nodeA.host.getOutboundStream(nodeB.host.id(), nodeB.host.protocolID)
require.Nil(t, stream, "node A should not have an outbound stream")
// node A opens the stream to send the first message
@@ -279,7 +279,7 @@ func TestExistingStream(t *testing.T) {
time.Sleep(TestMessageTimeout)
require.NotNil(t, handlerB.messages[nodeA.host.id()], "node B timeout waiting for message from node A")
- stream = nodeA.host.getStream(nodeB.host.id(), nodeB.host.protocolID)
+ stream = nodeA.host.getOutboundStream(nodeB.host.id(), nodeB.host.protocolID)
require.NotNil(t, stream, "node A should have an outbound stream")
// node A uses the stream to send a second message
@@ -287,7 +287,7 @@ func TestExistingStream(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, handlerB.messages[nodeA.host.id()], "node B timeout waiting for message from node A")
- stream = nodeA.host.getStream(nodeB.host.id(), nodeB.host.protocolID)
+ stream = nodeA.host.getOutboundStream(nodeB.host.id(), nodeB.host.protocolID)
require.NotNil(t, stream, "node B should have an outbound stream")
// node B opens the stream to send the first message
@@ -297,7 +297,7 @@ func TestExistingStream(t *testing.T) {
time.Sleep(TestMessageTimeout)
require.NotNil(t, handlerA.messages[nodeB.host.id()], "node A timeout waiting for message from node B")
- stream = nodeB.host.getStream(nodeA.host.id(), nodeB.host.protocolID)
+ stream = nodeB.host.getOutboundStream(nodeA.host.id(), nodeB.host.protocolID)
require.NotNil(t, stream, "node B should have an outbound stream")
// node B uses the stream to send a second message
@@ -305,7 +305,7 @@ func TestExistingStream(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, handlerA.messages[nodeB.host.id()], "node A timeout waiting for message from node B")
- stream = nodeB.host.getStream(nodeA.host.id(), nodeB.host.protocolID)
+ stream = nodeB.host.getOutboundStream(nodeA.host.id(), nodeB.host.protocolID)
require.NotNil(t, stream, "node B should have an outbound stream")
}
diff --git a/dot/network/notifications.go b/dot/network/notifications.go
index a733d06b78..98760d6e18 100644
--- a/dot/network/notifications.go
+++ b/dot/network/notifications.go
@@ -106,12 +106,12 @@ func (s *Service) createNotificationsMessageHandler(info *notificationsProtocol,
return errors.New("message is not NotificationsMessage")
}
- logger.Trace("received message on notifications sub-protocol", "protocol", info.protocolID,
- "message", msg,
- "peer", stream.Conn().RemotePeer(),
- )
-
if msg.IsHandshake() {
+ logger.Trace("received handshake on notifications sub-protocol", "protocol", info.protocolID,
+ "message", msg,
+ "peer", stream.Conn().RemotePeer(),
+ )
+
hs, ok := msg.(Handshake)
if !ok {
return errors.New("failed to convert message to Handshake")
@@ -186,6 +186,11 @@ func (s *Service) createNotificationsMessageHandler(info *notificationsProtocol,
return nil
}
+ logger.Debug("received message on notifications sub-protocol", "protocol", info.protocolID,
+ "message", msg,
+ "peer", stream.Conn().RemotePeer(),
+ )
+
err := messageHandler(peer, msg)
if err != nil {
return err
diff --git a/dot/network/service.go b/dot/network/service.go
index a970812985..6c454d611a 100644
--- a/dot/network/service.go
+++ b/dot/network/service.go
@@ -304,7 +304,7 @@ func (s *Service) handleConn(conn libp2pnetwork.Conn) {
defer info.mapMu.RUnlock()
peer := conn.RemotePeer()
- if hsData, has := info.getHandshakeData(peer); !has || !hsData.received {
+ if hsData, has := info.getHandshakeData(peer); !has || !hsData.received { //nolint
info.handshakeData.Store(peer, &handshakeData{
validated: false,
})
@@ -428,6 +428,9 @@ func (s *Service) RegisterNotificationsProtocol(sub protocol.ID,
info := s.notificationsProtocols[messageID]
+ decoder := createDecoder(info, handshakeDecoder, messageDecoder)
+ handlerWithValidate := s.createNotificationsMessageHandler(info, handshakeValidator, messageHandler)
+
s.host.registerStreamHandlerWithOverwrite(sub, overwriteProtocol, func(stream libp2pnetwork.Stream) {
logger.Trace("received stream", "sub-protocol", sub)
conn := stream.Conn()
@@ -437,10 +440,6 @@ func (s *Service) RegisterNotificationsProtocol(sub protocol.ID,
}
p := conn.RemotePeer()
-
- decoder := createDecoder(info, handshakeDecoder, messageDecoder)
- handlerWithValidate := s.createNotificationsMessageHandler(info, handshakeValidator, messageHandler)
-
s.readStream(stream, p, decoder, handlerWithValidate)
})
@@ -537,7 +536,7 @@ func (s *Service) readStream(stream libp2pnetwork.Stream, peer peer.ID, decoder
// decode message based on message type
msg, err := decoder(msgBytes[:tot], peer)
if err != nil {
- logger.Trace("Failed to decode message from peer", "peer", peer, "err", err)
+ logger.Trace("failed to decode message from peer", "protocol", stream.Protocol(), "err", err)
continue
}
diff --git a/dot/network/state.go b/dot/network/state.go
index 77725bd7de..72a4831e35 100644
--- a/dot/network/state.go
+++ b/dot/network/state.go
@@ -30,6 +30,7 @@ type BlockState interface {
GenesisHash() common.Hash
HasBlockBody(common.Hash) (bool, error)
GetFinalizedHeader(round, setID uint64) (*types.Header, error)
+ GetHashByNumber(num *big.Int) (common.Hash, error)
}
// Syncer is implemented by the syncing service
diff --git a/dot/network/state_test.go b/dot/network/state_test.go
index abc41d7a62..a268b9914e 100644
--- a/dot/network/state_test.go
+++ b/dot/network/state_test.go
@@ -75,3 +75,7 @@ func (mbs *MockBlockState) HasBlockBody(common.Hash) (bool, error) {
func (mbs *MockBlockState) GetFinalizedHeader(_, _ uint64) (*types.Header, error) {
return mbs.BestBlockHeader()
}
+
+func (mbs *MockBlockState) GetHashByNumber(_ *big.Int) (common.Hash, error) {
+ return common.Hash{}, nil
+}
diff --git a/dot/network/sync.go b/dot/network/sync.go
index 70d822d577..9cbc5d21e0 100644
--- a/dot/network/sync.go
+++ b/dot/network/sync.go
@@ -85,12 +85,20 @@ func (s *Service) handleSyncMessage(stream libp2pnetwork.Stream, msg Message) er
}
var (
- blockRequestSize uint32 = 128
- blockRequestBufferSize int = 6
+ blockRequestSize uint32 = 128
+ blockRequestBufferSize int = 6
+ blockResponseBufferSize int = 6
maxBlockResponseSize uint64 = 1024 * 1024 * 4 // 4mb
badPeerThreshold int = -2
protectedPeerThreshold int = 7
+
+ defaultSlotDuration = time.Second * 6
+)
+
+var (
+ errEmptyResponseData = fmt.Errorf("response data is empty")
+ errEmptyJustificationData = fmt.Errorf("no justifications in response data")
)
type syncPeer struct {
@@ -110,13 +118,15 @@ type requestData struct {
}
type syncQueue struct {
- s *Service
- ctx context.Context
- cancel context.CancelFunc
- peerScore *sync.Map // map[peer.ID]int; peers we have successfully synced from before -> their score; score increases on successful response
+ s *Service
+ slotDuration time.Duration
+ ctx context.Context
+ cancel context.CancelFunc
+ peerScore *sync.Map // map[peer.ID]int; peers we have successfully synced from before -> their score; score increases on successful response
- requestData *sync.Map // map[uint64]requestData; map of start # of request -> requestData
- requestCh chan *syncRequest
+ requestData *sync.Map // map[uint64]requestData; map of start # of request -> requestData
+ justificationRequestData *sync.Map // map[common.Hash]requestData; map of requests of justifications -> requestData
+ requestCh chan *syncRequest
responses []*types.BlockData
responseCh chan []*types.BlockData
@@ -133,22 +143,25 @@ func newSyncQueue(s *Service) *syncQueue {
ctx, cancel := context.WithCancel(s.ctx)
return &syncQueue{
- s: s,
- ctx: ctx,
- cancel: cancel,
- peerScore: new(sync.Map),
- requestData: new(sync.Map),
- requestCh: make(chan *syncRequest, blockRequestBufferSize),
- responses: []*types.BlockData{},
- responseCh: make(chan []*types.BlockData),
- benchmarker: newSyncBenchmarker(),
- buf: make([]byte, maxBlockResponseSize),
+ s: s,
+ slotDuration: defaultSlotDuration,
+ ctx: ctx,
+ cancel: cancel,
+ peerScore: new(sync.Map),
+ requestData: new(sync.Map),
+ justificationRequestData: new(sync.Map),
+ requestCh: make(chan *syncRequest, blockRequestBufferSize),
+ responses: []*types.BlockData{},
+ responseCh: make(chan []*types.BlockData, blockResponseBufferSize),
+ benchmarker: newSyncBenchmarker(),
+ buf: make([]byte, maxBlockResponseSize),
}
}
func (q *syncQueue) start() {
go q.handleResponseQueue()
go q.syncAtHead()
+ go q.finalizeAtHead()
go q.processBlockRequests()
go q.processBlockResponses()
@@ -169,7 +182,7 @@ func (q *syncQueue) syncAtHead() {
for {
select {
// sleep for average block time TODO: make this configurable from slot duration
- case <-time.After(time.Second * 6):
+ case <-time.After(q.slotDuration):
case <-q.ctx.Done():
return
}
@@ -180,7 +193,7 @@ func (q *syncQueue) syncAtHead() {
}
// we aren't at the head yet, sleep
- if curr.Number.Int64() < q.goal {
+ if curr.Number.Int64() < q.goal && curr.Number.Cmp(prev.Number) > 0 {
prev = curr
continue
}
@@ -395,17 +408,7 @@ func (q *syncQueue) pushRequest(start uint64, numRequests int, to peer.ID) {
start := best.Int64() + 1
req := createBlockRequest(start, 0)
- if d, has := q.requestData.Load(start); has {
- data := d.(requestData)
- // we haven't sent the request out yet, or we've already gotten the response
- if !data.sent || data.sent && data.received {
- logger.Debug("ignoring request, already received data", "start", start)
- return
- }
- }
-
logger.Debug("pushing request to queue", "start", start)
-
q.requestData.Store(start, requestData{
received: false,
})
@@ -453,7 +456,35 @@ func (q *syncQueue) pushRequest(start uint64, numRequests int, to peer.ID) {
func (q *syncQueue) pushResponse(resp *BlockResponseMessage, pid peer.ID) error {
if len(resp.BlockData) == 0 {
- return fmt.Errorf("response data is empty")
+ return errEmptyResponseData
+ }
+
+ startHash := resp.BlockData[0].Hash
+ if _, has := q.justificationRequestData.Load(startHash); has && !resp.BlockData[0].Header.Exists() {
+ numJustifications := 0
+ justificationResponses := []*types.BlockData{}
+
+ for _, bd := range resp.BlockData {
+ if bd.Justification.Exists() {
+ justificationResponses = append(justificationResponses, bd)
+ numJustifications++
+ }
+ }
+
+ if numJustifications == 0 {
+ return errEmptyJustificationData
+ }
+
+ q.updatePeerScore(pid, 1)
+ q.justificationRequestData.Store(startHash, requestData{
+ sent: true,
+ received: true,
+ from: pid,
+ })
+
+ logger.Info("pushed justification data to queue", "hash", startHash)
+ q.responseCh <- justificationResponses
+ return nil
}
start, end, err := resp.getStartAndEnd()
@@ -525,7 +556,7 @@ func (q *syncQueue) trySync(req *syncRequest) {
return
}
- logger.Debug("beginning to send out request", "start", req.req.StartingBlock.Value())
+ logger.Trace("beginning to send out request", "start", req.req.StartingBlock.Value())
if len(req.to) != 0 {
resp, err := q.syncWithPeer(req.to, req.req)
if err == nil {
@@ -535,11 +566,11 @@ func (q *syncQueue) trySync(req *syncRequest) {
}
}
- logger.Debug("failed to sync with peer", "peer", req.to, "error", err)
+ logger.Trace("failed to sync with peer", "peer", req.to, "error", err)
q.updatePeerScore(req.to, -1)
}
- logger.Debug("trying peers in prioritized order...")
+ logger.Trace("trying peers in prioritized order...")
syncPeers := q.getSortedPeers()
for _, peer := range syncPeers {
@@ -556,19 +587,24 @@ func (q *syncQueue) trySync(req *syncRequest) {
}
err = q.pushResponse(resp, peer.pid)
- if err != nil {
- logger.Trace("failed to push block response", "error", err)
+ if err != nil && err != errEmptyResponseData && err != errEmptyJustificationData {
+ logger.Debug("failed to push block response", "error", err)
} else {
return
}
}
- logger.Debug("failed to sync with any peer :(")
- if req.req.StartingBlock.IsUint64() {
+ logger.Trace("failed to sync with any peer :(")
+ if req.req.StartingBlock.IsUint64() && (req.req.RequestedData&RequestedDataHeader) == 1 {
q.requestData.Store(req.req.StartingBlock.Uint64(), requestData{
sent: true,
received: false,
})
+ } else if req.req.StartingBlock.IsHash() && (req.req.RequestedData&RequestedDataHeader) == 0 {
+ q.justificationRequestData.Store(req.req.StartingBlock.Hash(), requestData{
+ sent: true,
+ received: false,
+ })
}
req.to = ""
@@ -613,6 +649,12 @@ func (q *syncQueue) processBlockResponses() {
for {
select {
case data := <-q.responseCh:
+ // if the response doesn't contain a header, then it's a justification-only response
+ if !data[0].Header.Exists() {
+ q.handleBlockJustification(data)
+ continue
+ }
+
q.handleBlockData(data)
case <-q.ctx.Done():
return
@@ -620,6 +662,32 @@ func (q *syncQueue) processBlockResponses() {
}
}
+func (q *syncQueue) handleBlockJustification(data []*types.BlockData) {
+ startHash, endHash := data[0].Hash, data[len(data)-1].Hash
+ logger.Debug("sending justification data to syncer", "start", startHash, "end", endHash)
+
+ _, err := q.s.syncer.ProcessBlockData(data)
+ if err != nil {
+ logger.Warn("failed to handle block justifications", "error", err)
+ return
+ }
+
+ logger.Debug("finished processing justification data", "start", startHash, "end", endHash)
+
+ // update peer's score
+ var from peer.ID
+
+ d, ok := q.justificationRequestData.Load(startHash)
+ if !ok {
+ // this shouldn't happen
+ logger.Debug("can't find request data for response!", "start", startHash)
+ } else {
+ from = d.(requestData).from
+ q.updatePeerScore(from, 2)
+ q.justificationRequestData.Delete(startHash)
+ }
+}
+
func (q *syncQueue) handleBlockData(data []*types.BlockData) {
bestNum, err := q.s.blockState.BestBlockNumber()
if err != nil {
diff --git a/dot/network/sync_justification.go b/dot/network/sync_justification.go
new file mode 100644
index 0000000000..c5310cc144
--- /dev/null
+++ b/dot/network/sync_justification.go
@@ -0,0 +1,89 @@
+// Copyright 2019 ChainSafe Systems (ON) Corp.
+// This file is part of gossamer.
+//
+// The gossamer library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The gossamer library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the gossamer library. If not, see .
+
+package network
+
+import (
+ "math/big"
+ "time"
+)
+
+func (q *syncQueue) finalizeAtHead() {
+ prev, err := q.s.blockState.GetFinalizedHeader(0, 0)
+ if err != nil {
+ logger.Error("failed to get latest finalized block header", "error", err)
+ return
+ }
+
+ for {
+ select {
+ // sleep for average block time TODO: make this configurable from slot duration
+ case <-time.After(q.slotDuration * 2):
+ case <-q.ctx.Done():
+ return
+ }
+
+ curr, err := q.s.blockState.GetFinalizedHeader(0, 0)
+ if err != nil {
+ continue
+ }
+
+ logger.Debug("checking finalized blocks", "curr", curr.Number, "prev", prev.Number)
+
+ if curr.Number.Cmp(prev.Number) > 0 {
+ prev = curr
+ continue
+ }
+
+ prev = curr
+
+ // no new blocks have been finalized, request block justifications from peers
+ head, err := q.s.blockState.BestBlockNumber()
+ if err != nil {
+ continue
+ }
+
+ start := head.Uint64() - uint64(blockRequestSize)
+ if curr.Number.Uint64() > start {
+ start = curr.Number.Uint64() + 1
+ } else if int(start) < int(blockRequestSize) {
+ start = 1
+ }
+
+ q.pushJustificationRequest(start)
+ }
+}
+
+func (q *syncQueue) pushJustificationRequest(start uint64) {
+ startHash, err := q.s.blockState.GetHashByNumber(big.NewInt(int64(start)))
+ if err != nil {
+ logger.Error("failed to get hash for block w/ number", "number", start, "error", err)
+ return
+ }
+
+ req := createBlockRequestWithHash(startHash, blockRequestSize)
+ req.RequestedData = RequestedDataJustification
+
+ logger.Debug("pushing justification request to queue", "start", start, "hash", startHash)
+ q.justificationRequestData.Store(startHash, requestData{
+ received: false,
+ })
+
+ q.requestCh <- &syncRequest{
+ req: req,
+ to: "",
+ }
+}
diff --git a/dot/network/sync_justification_test.go b/dot/network/sync_justification_test.go
new file mode 100644
index 0000000000..00d854471a
--- /dev/null
+++ b/dot/network/sync_justification_test.go
@@ -0,0 +1,168 @@
+// Copyright 2019 ChainSafe Systems (ON) Corp.
+// This file is part of gossamer.
+//
+// The gossamer library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The gossamer library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the gossamer library. If not, see .
+
+package network
+
+import (
+ "context"
+ "math/big"
+ "testing"
+ "time"
+
+ "github.com/ChainSafe/gossamer/dot/types"
+ "github.com/ChainSafe/gossamer/lib/common"
+ "github.com/ChainSafe/gossamer/lib/common/optional"
+ "github.com/ChainSafe/gossamer/lib/utils"
+
+ "github.com/libp2p/go-libp2p-core/peer"
+ "github.com/stretchr/testify/require"
+)
+
+func TestSyncQueue_PushResponse_Justification(t *testing.T) {
+ basePath := utils.NewTestBasePath(t, "nodeA")
+ config := &Config{
+ BasePath: basePath,
+ Port: 7001,
+ RandSeed: 1,
+ NoBootstrap: true,
+ NoMDNS: true,
+ }
+
+ s := createTestService(t, config)
+ s.syncQueue.stop()
+ time.Sleep(time.Second)
+
+ peerID := peer.ID("noot")
+ msg := &BlockResponseMessage{
+ BlockData: []*types.BlockData{},
+ }
+
+ for i := 0; i < int(blockRequestSize); i++ {
+ msg.BlockData = append(msg.BlockData, &types.BlockData{
+ Hash: common.Hash{byte(i)},
+ Justification: optional.NewBytes(true, []byte{1}),
+ })
+ }
+
+ s.syncQueue.justificationRequestData.Store(common.Hash{byte(0)}, requestData{})
+ err := s.syncQueue.pushResponse(msg, peerID)
+ require.NoError(t, err)
+ require.Equal(t, 1, len(s.syncQueue.responseCh))
+ data, ok := s.syncQueue.justificationRequestData.Load(common.Hash{byte(0)})
+ require.True(t, ok)
+ require.Equal(t, requestData{
+ sent: true,
+ received: true,
+ from: peerID,
+ }, data)
+}
+
+func TestSyncQueue_PushResponse_EmptyJustification(t *testing.T) {
+ basePath := utils.NewTestBasePath(t, "nodeA")
+ config := &Config{
+ BasePath: basePath,
+ Port: 7001,
+ RandSeed: 1,
+ NoBootstrap: true,
+ NoMDNS: true,
+ }
+
+ s := createTestService(t, config)
+ s.syncQueue.stop()
+ time.Sleep(time.Second)
+
+ peerID := peer.ID("noot")
+ msg := &BlockResponseMessage{
+ BlockData: []*types.BlockData{},
+ }
+
+ for i := 0; i < int(blockRequestSize); i++ {
+ msg.BlockData = append(msg.BlockData, &types.BlockData{
+ Hash: common.Hash{byte(i)},
+ Justification: optional.NewBytes(false, nil),
+ })
+ }
+
+ s.syncQueue.justificationRequestData.Store(common.Hash{byte(0)}, &requestData{})
+ err := s.syncQueue.pushResponse(msg, peerID)
+ require.Equal(t, errEmptyJustificationData, err)
+}
+
+func TestSyncQueue_processBlockResponses_Justification(t *testing.T) {
+ q := newTestSyncQueue(t)
+ q.stop()
+ time.Sleep(time.Second)
+ q.ctx = context.Background()
+
+ go func() {
+ q.responseCh <- []*types.BlockData{
+ {
+ Hash: common.Hash{byte(0)},
+ Header: optional.NewHeader(false, nil),
+ Body: optional.NewBody(false, nil),
+ Receipt: optional.NewBytes(false, nil),
+ MessageQueue: optional.NewBytes(false, nil),
+ Justification: optional.NewBytes(true, []byte{1}),
+ },
+ }
+ }()
+
+ peerID := peer.ID("noot")
+ q.justificationRequestData.Store(common.Hash{byte(0)}, requestData{
+ from: peerID,
+ })
+
+ go q.processBlockResponses()
+ time.Sleep(time.Second)
+
+ _, has := q.justificationRequestData.Load(common.Hash{byte(0)})
+ require.False(t, has)
+
+ score, ok := q.peerScore.Load(peerID)
+ require.True(t, ok)
+ require.Equal(t, 2, score)
+}
+
+func TestSyncQueue_finalizeAtHead(t *testing.T) {
+ q := newTestSyncQueue(t)
+ q.stop()
+ time.Sleep(time.Second)
+ q.ctx = context.Background()
+ q.slotDuration = time.Millisecond * 200
+
+ hash, err := q.s.blockState.GetHashByNumber(big.NewInt(1))
+ require.NoError(t, err)
+
+ go q.finalizeAtHead()
+ time.Sleep(time.Second)
+
+ data, has := q.justificationRequestData.Load(hash)
+ require.True(t, has)
+ require.Equal(t, requestData{}, data)
+
+ expected := createBlockRequestWithHash(hash, blockRequestSize)
+ expected.RequestedData = RequestedDataJustification
+
+ select {
+ case req := <-q.requestCh:
+ require.Equal(t, &syncRequest{
+ req: expected,
+ to: "",
+ }, req)
+ case <-time.After(time.Second):
+ t.Fatal("did not receive request")
+ }
+}
diff --git a/dot/network/transaction.go b/dot/network/transaction.go
index 5eb1f7fa97..5131bc23ef 100644
--- a/dot/network/transaction.go
+++ b/dot/network/transaction.go
@@ -101,9 +101,7 @@ func (tm *TransactionMessage) IsHandshake() bool {
return false
}
-type transactionHandshake struct {
- Roles byte
-}
+type transactionHandshake struct{}
// SubProtocol returns the transactions sub-protocol
func (hs *transactionHandshake) SubProtocol() string {
@@ -112,22 +110,16 @@ func (hs *transactionHandshake) SubProtocol() string {
// String formats a transactionHandshake as a string
func (hs *transactionHandshake) String() string {
- return fmt.Sprintf("transactionHandshake Roles=%d",
- hs.Roles)
+ return "transactionHandshake"
}
// Encode encodes a transactionHandshake message using SCALE
func (hs *transactionHandshake) Encode() ([]byte, error) {
- return scale.Encode(hs)
+ return []byte{}, nil
}
// Decode the message into a transactionHandshake
func (hs *transactionHandshake) Decode(in []byte) error {
- msg, err := scale.Decode(in, hs)
- if err != nil {
- return err
- }
- hs.Roles = msg.(*transactionHandshake).Roles
return nil
}
@@ -147,19 +139,11 @@ func (hs *transactionHandshake) IsHandshake() bool {
}
func (s *Service) getTransactionHandshake() (Handshake, error) {
- return &transactionHandshake{
- Roles: s.cfg.Roles,
- }, nil
+ return &transactionHandshake{}, nil
}
func decodeTransactionHandshake(in []byte) (Handshake, error) {
- if len(in) < 1 {
- return nil, errors.New("invalid handshake")
- }
-
- return &transactionHandshake{
- Roles: in[0],
- }, nil
+ return &transactionHandshake{}, nil
}
func validateTransactionHandshake(_ peer.ID, _ Handshake) error {
diff --git a/dot/network/transaction_test.go b/dot/network/transaction_test.go
index db47f69e4b..07876168d9 100644
--- a/dot/network/transaction_test.go
+++ b/dot/network/transaction_test.go
@@ -29,9 +29,7 @@ import (
)
func TestDecodeTransactionHandshake(t *testing.T) {
- testHandshake := &transactionHandshake{
- Roles: 4,
- }
+ testHandshake := &transactionHandshake{}
enc, err := testHandshake.Encode()
require.NoError(t, err)
diff --git a/dot/network/utils.go b/dot/network/utils.go
index 49d3a9c054..62e9c53c00 100644
--- a/dot/network/utils.go
+++ b/dot/network/utils.go
@@ -185,11 +185,11 @@ func readStream(stream libp2pnetwork.Stream, buf []byte) (int, error) {
if err == io.EOF {
return 0, err
} else if err != nil {
- return 0, err // TODO: read bytes read from readLEB128ToUint64
+ return 0, err // TODO: return bytes read from readLEB128ToUint64
}
if length == 0 {
- return 0, err // TODO: read bytes read from readLEB128ToUint64
+ return 0, err // TODO: return bytes read from readLEB128ToUint64
}
// TODO: check if length > len(buf), if so probably log.Crit
diff --git a/dot/rpc/modules/system_test.go b/dot/rpc/modules/system_test.go
index 08e3512faf..c1f982530c 100644
--- a/dot/rpc/modules/system_test.go
+++ b/dot/rpc/modules/system_test.go
@@ -87,6 +87,10 @@ func (s *mockBlockState) GetFinalizedHeader(_, _ uint64) (*types.Header, error)
return s.BestBlockHeader()
}
+func (s *mockBlockState) GetHashByNumber(_ *big.Int) (common.Hash, error) {
+ return common.Hash{}, nil
+}
+
type mockTransactionHandler struct{}
func (h *mockTransactionHandler) HandleTransactionMessage(_ *network.TransactionMessage) error {
diff --git a/dot/state/block.go b/dot/state/block.go
index 7c2f33acd1..302d92fe5e 100644
--- a/dot/state/block.go
+++ b/dot/state/block.go
@@ -264,6 +264,16 @@ func (bs *BlockState) GetHeader(hash common.Hash) (*types.Header, error) {
return result, err
}
+// GetHashByNumber returns the block hash given the number
+func (bs *BlockState) GetHashByNumber(num *big.Int) (common.Hash, error) {
+ bh, err := bs.db.Get(headerHashKey(num.Uint64()))
+ if err != nil {
+ return common.Hash{}, fmt.Errorf("cannot get block %d: %s", num, err)
+ }
+
+ return common.NewHash(bh), nil
+}
+
// GetHeaderByNumber returns a block header given a number
func (bs *BlockState) GetHeaderByNumber(num *big.Int) (*types.Header, error) {
bh, err := bs.db.Get(headerHashKey(num.Uint64()))
diff --git a/dot/state/block_test.go b/dot/state/block_test.go
index 844e092d33..0a54db54a4 100644
--- a/dot/state/block_test.go
+++ b/dot/state/block_test.go
@@ -365,3 +365,29 @@ func TestFinalization_DeleteBlock(t *testing.T) {
// }
}
}
+
+func TestGetHashByNumber(t *testing.T) {
+ bs := newTestBlockState(t, testGenesisHeader)
+
+ res, err := bs.GetHashByNumber(big.NewInt(0))
+ require.NoError(t, err)
+ require.Equal(t, bs.genesisHash, res)
+
+ header := &types.Header{
+ Number: big.NewInt(1),
+ Digest: types.Digest{},
+ ParentHash: testGenesisHeader.Hash(),
+ }
+
+ block := &types.Block{
+ Header: header,
+ Body: &types.Body{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
+ }
+
+ err = bs.AddBlock(block)
+ require.NoError(t, err)
+
+ res, err = bs.GetHashByNumber(big.NewInt(1))
+ require.NoError(t, err)
+ require.Equal(t, header.Hash(), res)
+}
diff --git a/dot/sync/syncer.go b/dot/sync/syncer.go
index c5a96dd23c..b04aadb864 100644
--- a/dot/sync/syncer.go
+++ b/dot/sync/syncer.go
@@ -187,6 +187,11 @@ func (s *Service) ProcessBlockData(data []*types.BlockData) (int, error) {
logger.Debug("failed to add block to blocktree", "hash", bd.Hash, "error", err)
}
+ if bd.Justification != nil && bd.Justification.Exists() {
+ logger.Debug("handling Justification...", "number", header.Number, "hash", bd.Hash)
+ s.handleJustification(header, bd.Justification.Value())
+ }
+
continue
}
@@ -251,7 +256,7 @@ func (s *Service) ProcessBlockData(data []*types.BlockData) (int, error) {
logger.Debug("block processed", "hash", bd.Hash)
}
- if bd.Justification != nil && bd.Justification.Exists() {
+ if bd.Justification != nil && bd.Justification.Exists() && header != nil {
logger.Debug("handling Justification...", "number", bd.Number(), "hash", bd.Hash)
s.handleJustification(header, bd.Justification.Value())
}
diff --git a/lib/grandpa/network.go b/lib/grandpa/network.go
index 5b1b444638..f596282b5f 100644
--- a/lib/grandpa/network.go
+++ b/lib/grandpa/network.go
@@ -104,7 +104,7 @@ func (s *Service) registerProtocol() error {
func (s *Service) getHandshake() (Handshake, error) {
return &GrandpaHandshake{
- Roles: 0, // TODO: are roles returned?
+ Roles: 1, // TODO: don't hard-code this
}, nil
}
From f5de11ef4c833e29753ad614eb20aad4a020b93c Mon Sep 17 00:00:00 2001
From: noot <