Merge branch 'development'

Jiri Malek · Jiri Malek · commit a702dfa69171 · 2021-06-08T14:05:51.000+02:00
diff --git a/go.mod b/go.mod
@@ -19,7 +19,7 @@ require (
 	github.com/graph-gophers/graphql-transport-ws v0.0.0-20200904065757-c681d7e1b135
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/karalabe/usb v0.0.0-20210518091819-4ea20957c210 // indirect
-	github.com/klauspost/compress v1.11.12 // indirect
+	github.com/klauspost/compress v1.13.0 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/magiconair/properties v1.8.4 // indirect
 	github.com/mattn/go-runewidth v0.0.10 // indirect
diff --git a/go.sum b/go.sum
@@ -375,8 +375,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
 github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.9.5/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
-github.com/klauspost/compress v1.11.12 h1:famVnQVu7QwryBN4jNseQdUKES71ZAOnB6UQQJPZvqk=
-github.com/klauspost/compress v1.11.12/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/klauspost/compress v1.13.0 h1:2T7tUoQrQT+fQWdaY5rjWztFGAFwbGD04iPJg90ZiOs=
+github.com/klauspost/compress v1.13.0/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
 github.com/klauspost/cpuid v0.0.0-20170728055534-ae7887de9fa5/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/crc32 v0.0.0-20161016154125-cb6bfca970f6/go.mod h1:+ZoRqAPRLkC4NPOvfYeR5KNOrY6TD+/sAC3HXPZgDYg=
 github.com/klauspost/pgzip v1.0.2-0.20170402124221-0bf5dcad4ada/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
diff --git a/internal/config/config.go b/internal/config/config.go
@@ -108,6 +108,7 @@ type Database struct {
 // Cache represents the cache sub-system configuration.
 type Cache struct {
 	Eviction time.Duration `mapstructure:"eviction"`
+	MaxSize  int           `mapstructure:"size"`
 }
 
 // Compiler represents the contract compilers configuration.
diff --git a/internal/config/default.go b/internal/config/default.go
@@ -53,6 +53,9 @@ const (
 	// defCacheEvictionTime holds default time for in-memory eviction periods
 	defCacheEvictionTime = 15 * time.Minute
 
+	// defCacheMax size represents the default max size of the cache in MB
+	defCacheMaxSize = 4096
+
 	// defSolCompilerPath represents the default SOL compiler path
 	defSolCompilerPath = "/usr/bin/solc"
 
@@ -108,13 +111,16 @@ func applyDefaults(cfg *viper.Viper) {
 	cfg.SetDefault(keyLachesisUrl, defLachesisUrl)
 	cfg.SetDefault(keyMongoUrl, defMongoUrl)
 	cfg.SetDefault(keyMongoDatabase, defMongoDatabase)
-	cfg.SetDefault(keyCacheEvictionTime, defCacheEvictionTime)
 	cfg.SetDefault(keySolCompilerPath, defSolCompilerPath)
 	cfg.SetDefault(keyApiPeers, defApiPeers)
 	cfg.SetDefault(keyApiStateOrigin, defApiStateOrigin)
 	cfg.SetDefault(keyErc20TokenMapFilePath, defTokenLogoFilePath)
 	cfg.SetDefault(keyErc20Logos, defERC20Logo)
 
+	// in-memory cache
+	cfg.SetDefault(keyCacheEvictionTime, defCacheEvictionTime)
+	cfg.SetDefault(keyCacheMaxSize, defCacheMaxSize)
+
 	// server timeouts
 	cfg.SetDefault(keyTimeoutRead, defReadTimeout)
 	cfg.SetDefault(keyTimeoutWrite, defWriteTimeout)
diff --git a/internal/config/keys.go b/internal/config/keys.go
@@ -44,6 +44,7 @@ const (
 
 	// cache related options
 	keyCacheEvictionTime = "cache.eviction"
+	keyCacheMaxSize      = "cache.size"
 
 	// contract validation related
 	keySolCompilerPath = "compiler.sol"
diff --git a/internal/repository/cache/bridge.go b/internal/repository/cache/bridge.go
@@ -77,7 +77,7 @@ func cacheConfig(cfg *config.Config, log logger.Logger) bigcache.Config {
 		// cache will not allocate more memory than this limit, value in MB
 		// if value is reached then the oldest entries can be overridden for the new ones
 		// 0 value means no size limit
-		HardMaxCacheSize: 2048,
+		HardMaxCacheSize: cfg.Cache.MaxSize,
 
 		// callback fired when the oldest entry is removed because of its expiration time or no space left
 		// for the new entry, or because delete was called. A bitmask representing the reason will be returned.
diff --git a/internal/repository/cache/delegation.go b/internal/repository/cache/delegation.go
@@ -0,0 +1,57 @@
+// Package cache implements bridge to fast in-memory object cache.
+package cache
+
+import (
+	"fantom-api-graphql/internal/types"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"strings"
+)
+
+// delegationCacheKey generates cache key for the given delegation.
+func delegationCacheKey(adr common.Address, valID *hexutil.Big) string {
+	var key strings.Builder
+	key.WriteString("dlg")
+	key.WriteString(adr.String())
+	key.WriteString("to")
+	key.WriteString(valID.String())
+	return key.String()
+}
+
+// PullDelegation tries to pull delegation from the given address to the given validator
+// from internal in-memory cache.
+func (b *MemBridge) PullDelegation(adr common.Address, valID *hexutil.Big) *types.Delegation {
+	// try to get the account data from the cache
+	data, err := b.cache.Get(delegationCacheKey(adr, valID))
+	if err != nil {
+		return nil
+	}
+
+	// do we have the data?
+	dlg := new(types.Delegation)
+	if err := dlg.UnmarshalBSON(data); err != nil {
+		b.log.Criticalf("can not decode delegation data from in-memory cache; %s", err.Error())
+		return nil
+	}
+	return dlg
+}
+
+// PushDelegation stored the given delegation in memory cache.
+func (b *MemBridge) PushDelegation(dlg *types.Delegation) {
+	// no need to store nil
+	if dlg == nil {
+		return
+	}
+
+	// encode account
+	data, err := dlg.MarshalBSON()
+	if err != nil {
+		b.log.Criticalf("can not marshal delegation of %s to #%d; %s", dlg.Address.String(), dlg.ToStakerId.ToInt().Uint64(), err.Error())
+		return
+	}
+
+	// set the data to cache by block number
+	if err := b.cache.Set(delegationCacheKey(dlg.Address, dlg.ToStakerId), data); err != nil {
+		b.log.Criticalf("can not cache delegation of %s to #%d; %s", dlg.Address.String(), dlg.ToStakerId.ToInt().Uint64(), err.Error())
+	}
+}
diff --git a/internal/repository/cache/transaction.go b/internal/repository/cache/transaction.go
@@ -4,6 +4,7 @@ package cache
 import (
 	"fantom-api-graphql/internal/types"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/klauspost/compress/s2"
 )
 
 // PullTransaction extracts transaction information from the in-memory cache if available.
@@ -15,6 +16,12 @@ func (b *MemBridge) PullTransaction(hash *common.Hash) *types.Transaction {
 		return nil
 	}
 
+	// decode compressed transaction data from Snappy S2
+	data, err = s2.Decode(nil, data)
+	if err != nil {
+		return nil
+	}
+
 	// do we have the data?
 	trx := new(types.Transaction)
 	if err := trx.UnmarshalBSON(data); err != nil {
@@ -39,8 +46,15 @@ func (b *MemBridge) PushTransaction(trx *types.Transaction) {
 		return
 	}
 
+	// recover from s2 encoder panic
+	defer func() {
+		if r := recover(); r != nil {
+			b.log.Criticalf("can not encode transaction")
+		}
+	}()
+
 	// set the data to cache by block number
-	if err := b.cache.Set(trx.Hash.String(), data); err != nil {
+	if err := b.cache.Set(trx.Hash.String(), s2.Encode(nil, data)); err != nil {
 		b.log.Criticalf("can not cache transaction %s; %s", trx.Hash.String(), err.Error())
 	}
 }
diff --git a/internal/repository/db/account.go b/internal/repository/db/account.go
@@ -48,6 +48,12 @@ type AccountRow struct {
 	ScHash   *common.Hash `bson:"-"`
 }
 
+// initAccountsCollection initializes the account collection with
+// indexes and additional parameters needed by the app.
+func (db *MongoDbBridge) initAccountsCollection() {
+	db.log.Debugf("accounts collection initialized")
+}
+
 // Account tries to load an account identified by the address given from
 // the off-chain database.
 func (db *MongoDbBridge) Account(addr *common.Address) (*types.Account, error) {
@@ -91,12 +97,6 @@ func (db *MongoDbBridge) Account(addr *common.Address) (*types.Account, error) {
 	}, nil
 }
 
-// initAccountsCollection initializes the account collection with
-// indexes and additional parameters needed by the app.
-func (db *MongoDbBridge) initAccountsCollection() {
-	db.log.Debugf("accounts collection initialized")
-}
-
 // AddAccount stores an account in the blockchain if not exists.
 func (db *MongoDbBridge) AddAccount(acc *types.Account) error {
 	// do we have account data?
diff --git a/internal/repository/db/delegation.go b/internal/repository/db/delegation.go
@@ -60,6 +60,11 @@ func (db *MongoDbBridge) Delegation(addr *common.Address, valID *hexutil.Big) (*
 		{types.FiDelegationToValidator, valID.String()},
 	})
 
+	// do we have the data?
+	if sr.Err() != nil {
+		return nil, sr.Err()
+	}
+
 	// try to decode
 	var dlg types.Delegation
 	if err := sr.Decode(&dlg); err != nil {
diff --git a/internal/repository/repository.go b/internal/repository/repository.go
@@ -597,9 +597,6 @@ type proxy struct {
 
 	// service orchestrator reference
 	orc *orchestrator
-
-	// map of delegation balances for updates control
-	stakedAmounts *StakeAmountMap
 }
 
 // newRepository creates new instance of Repository implementation, namely proxy structure.
@@ -624,9 +621,6 @@ func newRepository() Repository {
 
 		// keep reference to the SOL compiler
 		solCompiler: cfg.Compiler.DefaultSolCompilerPath,
-
-		// map of delegation amounts
-		stakedAmounts: NewStakeAmountMap(),
 	}
 
 	// make the service orchestrator and start it's job
diff --git a/internal/repository/sfc_delegation.go b/internal/repository/sfc_delegation.go
@@ -38,37 +38,73 @@ func (p *proxy) StoreDelegation(dl *types.Delegation) error {
 }
 
 // UpdateDelegationBalance updates active balance of the given delegation.
-func (p *proxy) UpdateDelegationBalance(addr *common.Address, valID *hexutil.Big, unknown func(*big.Int) error) error {
+func (p *proxy) UpdateDelegationBalance(addr *common.Address, valID *hexutil.Big, unknownDelegation func(*big.Int) error) error {
 	// pull the current value
 	val, err := p.DelegationAmountStaked(addr, valID)
 	if err != nil {
 		p.log.Errorf("delegation balance not available for %s to %d; %s", addr.String(), valID.ToInt().Uint64(), err.Error())
 		return err
 	}
 
-	// check if we need to update it
-	if !p.stakedAmounts.Update(addr, valID, val) {
-		return nil
-	}
-
 	// do the update
-	err = p.db.UpdateDelegationBalance(addr, valID, (*hexutil.Big)(val))
+	err = p.updateDelegationBalance(addr, valID, val)
 	if err == nil {
 		return nil
 	}
 
 	// unknown delegation detected?
 	if err == db.ErrUnknownDelegation {
-		return unknown(val)
+		p.log.Errorf("delegation %s to %d not known yet", addr.String(), valID.ToInt().Uint64())
+		return unknownDelegation(val)
 	}
+
+	// some other error
 	p.log.Errorf("delegation %s to %d update failed; %s", addr.String(), valID.ToInt().Uint64(), err.Error())
 	return err
 }
 
+// updateDelegationBalance performs delegation balance update if needed.
+func (p *proxy) updateDelegationBalance(addr *common.Address, valID *hexutil.Big, amo *big.Int) error {
+	// get the delegation detail
+	dlg, err := p.Delegation(addr, valID)
+	if err != nil {
+		return err
+	}
+
+	// do we need to update? if the amount did not change, skip the update
+	if dlg.AmountStaked.ToInt().Cmp(amo) == 0 {
+		return nil
+	}
+
+	// update the delegation in DB and memory
+	dlg.AmountDelegated = (*hexutil.Big)(amo)
+	err = p.db.UpdateDelegationBalance(addr, valID, dlg.AmountDelegated)
+	if nil == err {
+		p.cache.PushDelegation(dlg)
+	}
+	return err
+}
+
 // Delegation returns a detail of delegation for the given address.
-func (p *proxy) Delegation(addr *common.Address, valID *hexutil.Big) (*types.Delegation, error) {
-	p.log.Debugf("loading delegation of %s to #%d", addr.String(), valID.ToInt().Uint64())
-	return p.db.Delegation(addr, valID)
+func (p *proxy) Delegation(adr *common.Address, valID *hexutil.Big) (*types.Delegation, error) {
+	// log what we do
+	p.log.Debugf("accessing delegation of %s to #%d", adr.String(), valID.ToInt().Uint64())
+
+	// try cache first
+	dlg := p.cache.PullDelegation(*adr, valID)
+	if dlg != nil {
+		return dlg, nil
+	}
+
+	// pull from DB instead; do we actually have it?
+	dlg, err := p.db.Delegation(adr, valID)
+	if err != nil {
+		return nil, err
+	}
+
+	// store to cache for future reference
+	p.cache.PushDelegation(dlg)
+	return dlg, nil
 }
 
 // DelegationAmountStaked returns the current amount of staked tokens for the given delegation.
diff --git a/internal/repository/stake_map.go b/internal/repository/stake_map.go
diff --git a/internal/types/delegation.go b/internal/types/delegation.go
@@ -31,9 +31,13 @@ type Delegation struct {
 	Address         common.Address `json:"address"`
 	ToStakerId      *hexutil.Big   `json:"toStakerID"`
 	ToStakerAddress common.Address `json:"toStakerAddr"`
-	AmountStaked    *hexutil.Big   `json:"amountStaked"`
-	AmountDelegated *hexutil.Big   `json:"amountDelegated"`
 	CreatedTime     hexutil.Uint64 `json:"createdTime"`
+
+	// AmountStaked represents the current staked amount
+	AmountStaked *hexutil.Big `json:"amountStaked"`
+
+	// AmountDelegated is the original amount delegated
+	AmountDelegated *hexutil.Big `json:"amountDelegated"`
 }
 
 // BsonDelegation represents the BSON i/o struct for a delegation.

Original file line number	Diff line number	Diff line change
`@@ -108,6 +108,7 @@ type Database struct {`
`108`	`108`	`// Cache represents the cache sub-system configuration.`
`109`	`109`	`type Cache struct {`
`110`	`110`	Eviction time.Duration `mapstructure:"eviction"`
	`111`	+ MaxSize int `mapstructure:"size"`
`111`	`112`	`}`
`112`	`113`
`113`	`114`	`// Compiler represents the contract compilers configuration.`