deep-stack/ipld-eth-server
4
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

major refactor pt 3

Browse Source
This commit is contained in:
Ian Norden
2020-06-29 19:16:52 -05:00
parent 449d23757e
commit e2bcc06f8a
49 changed files with 557 additions and 646 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/watch/service.go
+43 -42
View File
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package watcher
package watch
import (
"fmt"
@@ -22,13 +22,14 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/node"
ethnode "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/core"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/builders"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/node"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
@@ -37,12 +38,12 @@ const (
PayloadChanBufferSize = 2000
)
// SuperNode is the top level interface for streaming, converting to IPLDs, publishing,
// Watcher is the top level interface for streaming, converting to IPLDs, publishing,
// and indexing all chain data; screening this data; and serving it up to subscribed clients
// This service is compatible with the Ethereum service interface (node.Service)
type SuperNode interface {
type Watcher interface {
// APIs(), Protocols(), Start() and Stop()
node.Service
ethnode.Service
// Data processing event loop
Sync(wg *sync.WaitGroup, forwardPayloadChan chan<- shared.ConvertedData) error
// Pub-Sub handling event loop
@@ -52,12 +53,12 @@ type SuperNode interface {
// Method to unsubscribe from the service
Unsubscribe(id rpc.ID)
// Method to access the node info for the service
Node() *core.Node
Node() *node.Node
// Method to access chain type
Chain() shared.ChainType
}
// Service is the underlying struct for the super node
// Service is the underlying struct for the watcher
type Service struct {
// Used to sync access to the Subscriptions
sync.Mutex
@@ -83,8 +84,8 @@ type Service struct {
Subscriptions map[common.Hash]map[rpc.ID]Subscription
// A mapping of subscription params hash to the corresponding subscription params
SubscriptionTypes map[common.Hash]shared.SubscriptionSettings
// Info for the Geth node that this super node is working with
NodeInfo *core.Node
// Info for the Geth node that this watcher is working with
NodeInfo *node.Node
// Number of publishAndIndex workers
WorkerPoolSize int
// chain type for this service
@@ -97,40 +98,40 @@ type Service struct {
serveWg *sync.WaitGroup
}
// NewSuperNode creates a new super_node.Interface using an underlying super_node.Service struct
func NewSuperNode(settings *Config) (SuperNode, error) {
// NewWatcher creates a new Watcher using an underlying Service struct
func NewWatcher(settings *Config) (Watcher, error) {
sn := new(Service)
var err error
// If we are syncing, initialize the needed interfaces
if settings.Sync {
sn.Streamer, sn.PayloadChan, err = NewPayloadStreamer(settings.Chain, settings.WSClient)
sn.Streamer, sn.PayloadChan, err = builders.NewPayloadStreamer(settings.Chain, settings.WSClient)
if err != nil {
return nil, err
}
sn.Converter, err = NewPayloadConverter(settings.Chain)
sn.Converter, err = builders.NewPayloadConverter(settings.Chain)
if err != nil {
return nil, err
}
sn.Publisher, err = NewIPLDPublisher(settings.Chain, settings.IPFSPath, settings.SyncDBConn, settings.IPFSMode)
sn.Publisher, err = builders.NewIPLDPublisher(settings.Chain, settings.IPFSPath, settings.SyncDBConn, settings.IPFSMode)
if err != nil {
return nil, err
}
sn.Indexer, err = NewCIDIndexer(settings.Chain, settings.SyncDBConn, settings.IPFSMode)
sn.Indexer, err = builders.NewCIDIndexer(settings.Chain, settings.SyncDBConn, settings.IPFSMode)
if err != nil {
return nil, err
}
sn.Filterer, err = NewResponseFilterer(settings.Chain)
sn.Filterer, err = builders.NewResponseFilterer(settings.Chain)
if err != nil {
return nil, err
}
}
// If we are serving, initialize the needed interfaces
if settings.Serve {
sn.Retriever, err = NewCIDRetriever(settings.Chain, settings.ServeDBConn)
sn.Retriever, err = builders.NewCIDRetriever(settings.Chain, settings.ServeDBConn)
if err != nil {
return nil, err
}
sn.IPLDFetcher, err = NewIPLDFetcher(settings.Chain, settings.IPFSPath, settings.ServeDBConn, settings.IPFSMode)
sn.IPLDFetcher, err = builders.NewIPLDFetcher(settings.Chain, settings.IPFSPath, settings.ServeDBConn, settings.IPFSMode)
if err != nil {
return nil, err
}
@@ -151,14 +152,14 @@ func (sap *Service) Protocols() []p2p.Protocol {
return []p2p.Protocol{}
}
// APIs returns the RPC descriptors the super node service offers
// APIs returns the RPC descriptors the watcher service offers
func (sap *Service) APIs() []rpc.API {
ifnoAPI := NewInfoAPI()
apis := []rpc.API{
{
Namespace: APIName,
Version: APIVersion,
Service: NewPublicSuperNodeAPI(sap),
Service: NewPublicWatcherAPI(sap),
Public: true,
},
{
@@ -180,7 +181,7 @@ func (sap *Service) APIs() []rpc.API {
Public: true,
},
}
chainAPI, err := NewPublicAPI(sap.chain, sap.db, sap.ipfsPath)
chainAPI, err := builders.NewPublicAPI(sap.chain, sap.db, sap.ipfsPath)
if err != nil {
log.Error(err)
return apis
@@ -211,7 +212,7 @@ func (sap *Service) Sync(wg *sync.WaitGroup, screenAndServePayload chan<- shared
case payload := <-sap.PayloadChan:
ipldPayload, err := sap.Converter.Convert(payload)
if err != nil {
log.Errorf("super node conversion error for chain %s: %v", sap.chain.String(), err)
log.Errorf("watcher conversion error for chain %s: %v", sap.chain.String(), err)
continue
}
log.Infof("%s data streamed at head height %d", sap.chain.String(), ipldPayload.Height())
@@ -229,7 +230,7 @@ func (sap *Service) Sync(wg *sync.WaitGroup, screenAndServePayload chan<- shared
publishAndIndexPayload <- ipldPayload
}
case err := <-sub.Err():
log.Errorf("super node subscription error for chain %s: %v", sap.chain.String(), err)
log.Errorf("watcher subscription error for chain %s: %v", sap.chain.String(), err)
case <-sap.QuitChan:
log.Infof("quiting %s Sync process", sap.chain.String())
return
@@ -248,18 +249,18 @@ func (sap *Service) publishAndIndex(wg *sync.WaitGroup, id int, publishAndIndexP
for {
select {
case payload := <-publishAndIndexPayload:
log.Debugf("%s super node publishAndIndex worker %d publishing data streamed at head height %d", sap.chain.String(), id, payload.Height())
log.Debugf("%s watcher publishAndIndex worker %d publishing data streamed at head height %d", sap.chain.String(), id, payload.Height())
cidPayload, err := sap.Publisher.Publish(payload)
if err != nil {
log.Errorf("%s super node publishAndIndex worker %d publishing error: %v", sap.chain.String(), id, err)
log.Errorf("%s watcher publishAndIndex worker %d publishing error: %v", sap.chain.String(), id, err)
continue
}
log.Debugf("%s super node publishAndIndex worker %d indexing data streamed at head height %d", sap.chain.String(), id, payload.Height())
log.Debugf("%s watcher publishAndIndex worker %d indexing data streamed at head height %d", sap.chain.String(), id, payload.Height())
if err := sap.Indexer.Index(cidPayload); err != nil {
log.Errorf("%s super node publishAndIndex worker %d indexing error: %v", sap.chain.String(), id, err)
log.Errorf("%s watcher publishAndIndex worker %d indexing error: %v", sap.chain.String(), id, err)
}
case <-sap.QuitChan:
log.Infof("%s super node publishAndIndex worker %d shutting down", sap.chain.String(), id)
log.Infof("%s watcher publishAndIndex worker %d shutting down", sap.chain.String(), id)
return
}
}
@@ -298,7 +299,7 @@ func (sap *Service) filterAndServe(payload shared.ConvertedData) {
// Retrieve the subscription parameters for this subscription type
subConfig, ok := sap.SubscriptionTypes[ty]
if !ok {
log.Errorf("super node %s subscription configuration for subscription type %s not available", sap.chain.String(), ty.Hex())
log.Errorf("watcher %s subscription configuration for subscription type %s not available", sap.chain.String(), ty.Hex())
sap.closeType(ty)
continue
}
@@ -310,19 +311,19 @@ func (sap *Service) filterAndServe(payload shared.ConvertedData) {
}
response, err := sap.Filterer.Filter(subConfig, payload)
if err != nil {
log.Errorf("super node filtering error for chain %s: %v", sap.chain.String(), err)
log.Errorf("watcher filtering error for chain %s: %v", sap.chain.String(), err)
sap.closeType(ty)
continue
}
responseRLP, err := rlp.EncodeToBytes(response)
if err != nil {
log.Errorf("super node rlp encoding error for chain %s: %v", sap.chain.String(), err)
log.Errorf("watcher rlp encoding error for chain %s: %v", sap.chain.String(), err)
continue
}
for id, sub := range subs {
select {
case sub.PayloadChan <- SubscriptionPayload{Data: responseRLP, Err: "", Flag: EmptyFlag, Height: response.Height()}:
log.Debugf("sending super node %s payload to subscription %s", sap.chain.String(), id)
log.Debugf("sending watcher %s payload to subscription %s", sap.chain.String(), id)
default:
log.Infof("unable to send %s payload to subscription %s; channel has no receiver", sap.chain.String(), id)
}
@@ -368,7 +369,7 @@ func (sap *Service) Subscribe(id rpc.ID, sub chan<- SubscriptionPayload, quitCha
// Otherwise we only filter new data as it is streamed in from the state diffing geth node
if params.HistoricalData() || params.HistoricalDataOnly() {
if err := sap.sendHistoricalData(subscription, id, params); err != nil {
sendNonBlockingErr(subscription, fmt.Errorf("%s super node subscriber backfill error: %v", sap.chain.String(), err))
sendNonBlockingErr(subscription, fmt.Errorf("%s watcher subscriber backfill error: %v", sap.chain.String(), err))
sendNonBlockingQuit(subscription)
return
}
@@ -404,13 +405,13 @@ func (sap *Service) sendHistoricalData(sub Subscription, id rpc.ID, params share
for i := startingBlock; i <= endingBlock; i++ {
select {
case <-sap.QuitChan:
log.Infof("%s super node historical data feed to subscription %s closed", sap.chain.String(), id)
log.Infof("%s watcher historical data feed to subscription %s closed", sap.chain.String(), id)
return
default:
}
cidWrappers, empty, err := sap.Retriever.Retrieve(params, i)
if err != nil {
sendNonBlockingErr(sub, fmt.Errorf(" %s super node CID Retrieval error at block %d\r%s", sap.chain.String(), i, err.Error()))
sendNonBlockingErr(sub, fmt.Errorf(" %s watcher CID Retrieval error at block %d\r%s", sap.chain.String(), i, err.Error()))
continue
}
if empty {
@@ -419,7 +420,7 @@ func (sap *Service) sendHistoricalData(sub Subscription, id rpc.ID, params share
for _, cids := range cidWrappers {
response, err := sap.IPLDFetcher.Fetch(cids)
if err != nil {
sendNonBlockingErr(sub, fmt.Errorf("%s super node IPLD Fetching error at block %d\r%s", sap.chain.String(), i, err.Error()))
sendNonBlockingErr(sub, fmt.Errorf("%s watcher IPLD Fetching error at block %d\r%s", sap.chain.String(), i, err.Error()))
continue
}
responseRLP, err := rlp.EncodeToBytes(response)
@@ -429,7 +430,7 @@ func (sap *Service) sendHistoricalData(sub Subscription, id rpc.ID, params share
}
select {
case sub.PayloadChan <- SubscriptionPayload{Data: responseRLP, Err: "", Flag: EmptyFlag, Height: response.Height()}:
log.Debugf("sending super node historical data payload to %s subscription %s", sap.chain.String(), id)
log.Debugf("sending watcher historical data payload to %s subscription %s", sap.chain.String(), id)
default:
log.Infof("unable to send backFill payload to %s subscription %s; channel has no receiver", sap.chain.String(), id)
}
@@ -448,7 +449,7 @@ func (sap *Service) sendHistoricalData(sub Subscription, id rpc.ID, params share
// Unsubscribe is used by the API to remotely unsubscribe to the StateDiffingService loop
func (sap *Service) Unsubscribe(id rpc.ID) {
log.Infof("Unsubscribing %s from the %s super node service", id, sap.chain.String())
log.Infof("Unsubscribing %s from the %s watcher service", id, sap.chain.String())
sap.Lock()
for ty := range sap.Subscriptions {
delete(sap.Subscriptions[ty], id)
@@ -464,7 +465,7 @@ func (sap *Service) Unsubscribe(id rpc.ID) {
// Start is used to begin the service
// This is mostly just to satisfy the node.Service interface
func (sap *Service) Start(*p2p.Server) error {
log.Infof("Starting %s super node service", sap.chain.String())
log.Infof("Starting %s watcher service", sap.chain.String())
wg := new(sync.WaitGroup)
payloadChan := make(chan shared.ConvertedData, PayloadChanBufferSize)
if err := sap.Sync(wg, payloadChan); err != nil {
@@ -477,7 +478,7 @@ func (sap *Service) Start(*p2p.Server) error {
// Stop is used to close down the service
// This is mostly just to satisfy the node.Service interface
func (sap *Service) Stop() error {
log.Infof("Stopping %s super node service", sap.chain.String())
log.Infof("Stopping %s watcher service", sap.chain.String())
sap.Lock()
close(sap.QuitChan)
sap.close()
@@ -486,7 +487,7 @@ func (sap *Service) Stop() error {
}
// Node returns the node info for this service
func (sap *Service) Node() *core.Node {
func (sap *Service) Node() *node.Node {
return sap.NodeInfo
}