cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

implement basic message filtering

Browse Source
This commit is contained in:
whyrusleeping 2020-02-27 17:39:07 -08:00
parent 0eeb6c16d9
commit 00efd097c7
3 changed files with 27 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
chain/sub/incoming.go
View File

@ -153,9 +153,32 @@ func (brc *blockReceiptCache) add(bcid cid.Cid) int {
return val.(int)
}
type MessageValidator struct {
mpool *messagepool.MessagePool
}
func NewMessageValidator(mp *messagepool.MessagePool) *MessageValidator {
return &MessageValidator{mp}
}
func (mv *MessageValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub.Message) bool {
m, err := types.DecodeSignedMessage(msg.Message.GetData())
if err != nil {
log.Warnf("failed to decode incoming message: %s", err)
return false
}
if err := mv.mpool.Add(m); err != nil {
log.Warnf("failed to add message from network to message pool (From: %s, To: %s, Nonce: %d, Value: %s): %s", m.Message.From, m.Message.To, m.Message.Nonce, types.FIL(m.Message.Value), err)
return false
}
return true
}
func HandleIncomingMessages(ctx context.Context, mpool *messagepool.MessagePool, msub *pubsub.Subscription) {
for {
msg, err := msub.Next(ctx)
_, err := msub.Next(ctx)
if err != nil {
log.Warn("error from message subscription: ", err)
if ctx.Err() != nil {
@ -165,15 +188,6 @@ func HandleIncomingMessages(ctx context.Context, mpool *messagepool.MessagePool,
continue
}
m, ok := msg.ValidatorData.(*types.SignedMessage)
if !ok {
log.Errorf("message validator func passed on wrong type: %#v", msg.ValidatorData)
continue
}
if err := mpool.Add(m); err != nil {
log.Warnf("failed to add message from network to message pool (From: %s, To: %s, Nonce: %d, Value: %s): %s", m.Message.From, m.Message.To, m.Message.Nonce, types.FIL(m.Message.Value), err)
continue
}
// Do nothing... everything happens in validate
}
}

1
go.mod
View File

@ -84,7 +84,6 @@ require (
github.com/multiformats/go-varint v0.0.2
github.com/opentracing/opentracing-go v1.1.0
github.com/polydawn/refmt v0.0.0-20190809202753-05966cbd336a
github.com/prometheus/client_golang v0.9.3-0.20190127221311-3c4408c8b829
github.com/prometheus/common v0.2.0
github.com/stretchr/testify v1.4.0
github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba

14
node/modules/services.go
View File

@ -16,7 +16,6 @@ import (
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/sub"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/hello"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
@ -78,18 +77,9 @@ func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub
panic(err)
}
v := func(ctx context.Context, pid peer.ID, msg *pubsub.Message) bool {
m, err := types.DecodeSignedMessage(msg.GetData())
if err != nil {
log.Errorf("got incorrectly formatted Message: %s", err)
return false
}
v := sub.NewMessageValidator(mpool)
msg.ValidatorData = m
return true
}
if err := ps.RegisterTopicValidator(MessagesTopic, v); err != nil {
if err := ps.RegisterTopicValidator(MessagesTopic, v.Validate); err != nil {
panic(err)
}