diff --git a/cmd/smoketest.go b/cmd/smoketest.go index 82e5e0ca69..8d784cf36a 100644 --- a/cmd/smoketest.go +++ b/cmd/smoketest.go @@ -42,6 +42,14 @@ const smokeTestDescription = `The smoke-test command creates a local threshold executed, once again with an in-process broadcast channel and chain, and the final signature is verified by each member of the group.` +type noopPersistence struct { +} + +func (np *noopPersistence) Save(data []byte, name string) error { + // noop + return nil +} + func init() { SmokeTestCommand = cli.Command{ Name: "smoke-test", @@ -140,6 +148,8 @@ func createNode( )) } + storage := &noopPersistence{} + netProvider := netlocal.Connect() go func() { @@ -157,6 +167,7 @@ func createNode( chainCounter, stakeMonitor, netProvider, + storage, ) if err != nil { panic(fmt.Sprintf( diff --git a/cmd/start.go b/cmd/start.go index de00a24aee..d22d95e6e5 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -11,6 +11,7 @@ import ( "github.com/keep-network/keep-core/pkg/net/key" "github.com/keep-network/keep-core/pkg/net/libp2p" "github.com/keep-network/keep-core/pkg/operator" + "github.com/keep-network/keep-core/pkg/persistence" "github.com/urfave/cli" ) @@ -100,6 +101,8 @@ func Start(c *cli.Context) error { isBootstrapNode := config.LibP2P.Seed != 0 nodeHeader(isBootstrapNode, netProvider.AddrStrings(), port) + persistence := persistence.NewDiskHandle(config.Storage.DataDir) + err = beacon.Initialize( ctx, config.Ethereum.Account.Address, @@ -107,6 +110,7 @@ func Start(c *cli.Context) error { blockCounter, stakeMonitor, netProvider, + persistence, ) if err != nil { return fmt.Errorf("error initializing beacon: [%v]", err) diff --git a/config/config_test.go b/config/config_test.go index eb72b7e4b3..aaca20e397 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -47,7 +47,7 @@ func TestReadConfig(t *testing.T) { "KeepGroup": "0xcf64c2a367341170cb4e09cf8c0ed137d8473ceb", }, }, - "StateManagementData": { + "Storage.DataDir": { readValueFunc: func(c *Config) interface{} { return c.Storage.DataDir }, expectedValue: "/my/secure/location", }, diff --git a/pkg/beacon/beacon.go b/pkg/beacon/beacon.go index 8124708efb..9af570ec08 100644 --- a/pkg/beacon/beacon.go +++ b/pkg/beacon/beacon.go @@ -11,6 +11,7 @@ import ( "github.com/keep-network/keep-core/pkg/beacon/relay/registry" "github.com/keep-network/keep-core/pkg/chain" "github.com/keep-network/keep-core/pkg/net" + "github.com/keep-network/keep-core/pkg/persistence" ) // Initialize kicks off the random beacon by initializing internal state, @@ -24,6 +25,7 @@ func Initialize( blockCounter chain.BlockCounter, stakeMonitor chain.StakeMonitor, netProvider net.Provider, + persistence persistence.Handle, ) error { chainConfig, err := relayChain.GetConfig() if err != nil { @@ -35,7 +37,7 @@ func Initialize( return err } - groupRegistry := registry.NewGroupRegistry(relayChain) + groupRegistry := registry.NewGroupRegistry(relayChain, persistence) node := relay.NewNode( staker, diff --git a/pkg/beacon/relay/node.go b/pkg/beacon/relay/node.go index ce55d1bb4b..6fbb0d855e 100644 --- a/pkg/beacon/relay/node.go +++ b/pkg/beacon/relay/node.go @@ -97,10 +97,13 @@ func (n *Node) JoinGroupIfEligible( return } - n.groupRegistry.RegisterGroup( + err = n.groupRegistry.RegisterGroup( signer, broadcastChannelName, ) + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to register a group: [%v].\n", err) + } }() } } diff --git a/pkg/beacon/relay/registry/groups.go b/pkg/beacon/relay/registry/groups.go index b4efb51b19..a7d81d6746 100644 --- a/pkg/beacon/relay/registry/groups.go +++ b/pkg/beacon/relay/registry/groups.go @@ -7,6 +7,8 @@ import ( relaychain "github.com/keep-network/keep-core/pkg/beacon/relay/chain" "github.com/keep-network/keep-core/pkg/beacon/relay/dkg" + + "github.com/keep-network/keep-core/pkg/persistence" ) // Groups represents a collection of Keep groups in which the given @@ -17,6 +19,8 @@ type Groups struct { myGroups map[string][]*Membership relayChain relaychain.GroupRegistrationInterface + + storage storage } // Membership represents a member of a group @@ -28,10 +32,13 @@ type Membership struct { // NewGroupRegistry returns an empty GroupRegistry. func NewGroupRegistry( relayChain relaychain.GroupRegistrationInterface, + persistence persistence.Handle, ) *Groups { return &Groups{ myGroups: make(map[string][]*Membership), relayChain: relayChain, + storage: newStorage(persistence), + mutex: sync.Mutex{}, } } @@ -40,18 +47,25 @@ func NewGroupRegistry( func (gr *Groups) RegisterGroup( signer *dkg.ThresholdSigner, channelName string, -) { - +) error { gr.mutex.Lock() defer gr.mutex.Unlock() groupPublicKey := string(signer.GroupPublicKeyBytes()) - gr.myGroups[groupPublicKey] = append(gr.myGroups[groupPublicKey], - &Membership{ - Signer: signer, - ChannelName: channelName, - }) + membership := &Membership{ + Signer: signer, + ChannelName: channelName, + } + + err := gr.storage.save(membership) + if err != nil { + return fmt.Errorf("could not persist membership to the storage: [%v]", err) + } + + gr.myGroups[groupPublicKey] = append(gr.myGroups[groupPublicKey], membership) + + return nil } // GetGroup gets a group by a groupPublicKey diff --git a/pkg/beacon/relay/registry/groups_test.go b/pkg/beacon/relay/registry/groups_test.go index cd8ce13965..4390aa36d9 100644 --- a/pkg/beacon/relay/registry/groups_test.go +++ b/pkg/beacon/relay/registry/groups_test.go @@ -3,7 +3,6 @@ package registry import ( "bytes" "math/big" - "sync" "testing" bn256 "github.com/ethereum/go-ethereum/crypto/bn256/cloudflare" @@ -14,19 +13,26 @@ import ( "github.com/keep-network/keep-core/pkg/subscription" ) +type noopPersistence struct { +} + +func (np *noopPersistence) Save([]byte, string) error { + // noop + return nil +} + func TestRegisterGroup(t *testing.T) { + noopPersistence := &noopPersistence{} + chain := chainLocal.Connect(5, 3, big.NewInt(200)).ThresholdRelay() + + gr := NewGroupRegistry(chain, noopPersistence) + signer := dkg.NewThresholdSigner( group.MemberIndex(2), new(bn256.G2).ScalarBaseMult(big.NewInt(10)), big.NewInt(1), ) - gr := &Groups{ - mutex: sync.Mutex{}, - myGroups: make(map[string][]*Membership), - relayChain: chainLocal.Connect(5, 3, big.NewInt(200)).ThresholdRelay(), - } - gr.RegisterGroup(signer, "test_channel") actual := gr.GetGroup(signer.GroupPublicKeyBytes()) @@ -50,12 +56,9 @@ func TestUnregisterStaleGroups(t *testing.T) { mockChain := &mockGroupRegistrationInterface{ groupsToRemove: [][]byte{}, } + noopPersistence := &noopPersistence{} - gr := &Groups{ - mutex: sync.Mutex{}, - myGroups: make(map[string][]*Membership), - relayChain: mockChain, - } + gr := NewGroupRegistry(mockChain, noopPersistence) signer1 := dkg.NewThresholdSigner( group.MemberIndex(1), diff --git a/pkg/beacon/relay/registry/storage.go b/pkg/beacon/relay/registry/storage.go new file mode 100644 index 0000000000..d96d09c796 --- /dev/null +++ b/pkg/beacon/relay/registry/storage.go @@ -0,0 +1,34 @@ +package registry + +import ( + "fmt" + + "github.com/keep-network/keep-core/pkg/persistence" + + "encoding/hex" +) + +type storage interface { + save(membership *Membership) error +} + +type persistentStorage struct { + handle persistence.Handle +} + +func newStorage(persistence persistence.Handle) storage { + return &persistentStorage{ + handle: persistence, + } +} + +// Save converts a membership suitable for disk storage. +func (ps *persistentStorage) save(membership *Membership) error { + membershipBytes, err := membership.Marshal() + if err != nil { + return fmt.Errorf("marshalling of the membership failed: [%v]", err) + } + hexGroupPublicKey := hex.EncodeToString(membership.Signer.GroupPublicKeyBytes()) + + return ps.handle.Save(membershipBytes, "/membership_"+hexGroupPublicKey+"_"+fmt.Sprint(membership.Signer.MemberID())) +} diff --git a/pkg/persistence/disk_persistence.go b/pkg/persistence/disk_persistence.go new file mode 100644 index 0000000000..bdbcea4f35 --- /dev/null +++ b/pkg/persistence/disk_persistence.go @@ -0,0 +1,97 @@ +package persistence + +import ( + "fmt" + "io/ioutil" + "os" +) + +// NewDiskHandle creates on-disk data persistence handle +func NewDiskHandle(path string) Handle { + return &diskPersistence{ + dataDir: path, + } +} + +type diskPersistence struct { + dataDir string +} + +// Save - writes data to file +func (ds *diskPersistence) Save(data []byte, suffix string) error { + file := &file{ + fileName: ds.dataDir + suffix, + } + + return file.write(data) +} + +var ( + //ErrNoFileExists an error is shown when no file name was provided + errNoFileExists = fmt.Errorf("please provide a file name") +) + +// File represents a file on disk that a caller can use to read and write into. +type file struct { + // FileName is the file name of the main storage file. + fileName string +} + +// Create and write data to a file +func (f *file) write(data []byte) error { + if f.fileName == "" { + return errNoFileExists + } + + var err error + writeFile, err := os.Create(f.fileName) + if err != nil { + return err + } + + defer writeFile.Close() + + _, err = writeFile.Write(data) + if err != nil { + return err + } + + writeFile.Sync() + + return nil +} + +// Read a file from a file system +func (f *file) read(fileName string) ([]byte, error) { + if f.fileName == "" { + return nil, errNoFileExists + } + + readFile, err := os.Open(fileName) + if err != nil { + return nil, err + } + + defer readFile.Close() + + data, err := ioutil.ReadAll(readFile) + if err != nil { + return nil, err + } + + return data, nil +} + +// Remove a file from a file system +func (f *file) remove(fileName string) error { + if f.fileName == "" { + return errNoFileExists + } + + err := os.Remove(fileName) + if err != nil { + return err + } + + return nil +} diff --git a/pkg/persistence/disk_persistence_test.go b/pkg/persistence/disk_persistence_test.go new file mode 100644 index 0000000000..cb3f6531c5 --- /dev/null +++ b/pkg/persistence/disk_persistence_test.go @@ -0,0 +1,49 @@ +package persistence + +import ( + "bytes" + "os" + "testing" +) + +var ( + fileName = "foo" +) + +func TestMain(m *testing.M) { + code := m.Run() + if _, err := os.Stat(fileName); err == nil { + os.Remove(fileName) + } + os.Exit(code) +} + +func TestFile_WriteRead(t *testing.T) { + file := &file{ + fileName: fileName, + } + bytesToTest := []byte{115, 111, 109, 101, 10} + + file.write(bytesToTest) + + actual, _ := file.read(fileName) + + if !bytes.Equal(bytesToTest, actual) { + t.Fatalf("Bytes do not match. \nExpected: [%+v]\nActual: [%+v]", + bytesToTest, + actual) + } +} + +func TestFile_Remove(t *testing.T) { + if _, err := os.Stat(fileName); err == nil { + err = os.Remove(fileName) + if err != nil { + t.Fatalf("Was not able to remove a file [%+v]", fileName) + } + } + + if _, err := os.Stat(fileName); err == nil { + t.Fatalf("File [%+v] was supposed to be removed", fileName) + } +} diff --git a/pkg/persistence/persistence.go b/pkg/persistence/persistence.go new file mode 100644 index 0000000000..f28e1c3c11 --- /dev/null +++ b/pkg/persistence/persistence.go @@ -0,0 +1,7 @@ +package persistence + +// Handle is an interface for data persistence. Underlying implementation +// can write data e.g. to disk, cache, or hardware module. +type Handle interface { + Save(data []byte, name string) error +}