feat(settler): monitor settled channels

Add a routine to monitor the chain for payment channels settling, and if it's a channel that we are
the recipient of payments on, submit any outstanding vouchers to the
chain

node/builder.go

@@ -57,6 +57,7 @@ import (
 	"github.com/filecoin-project/lotus/node/modules/testing"
 	"github.com/filecoin-project/lotus/node/repo"
 	"github.com/filecoin-project/lotus/paychmgr"
+	"github.com/filecoin-project/lotus/paychmgr/settler"
 	"github.com/filecoin-project/lotus/storage"
 	"github.com/filecoin-project/lotus/storage/sectorblocks"
 	sectorstorage "github.com/filecoin-project/sector-storage"
@@ -118,6 +119,7 @@ const (
 	// daemon
 	ExtractApiKey
 	HeadMetricsKey
+	SettlePaymentChannelsKey
 	RunPeerTaggerKey
 	JournalKey
 
@@ -272,6 +274,7 @@ func Online() Option {
 			Override(new(*paychmgr.Store), paychmgr.NewStore),
 			Override(new(*paychmgr.Manager), paychmgr.NewManager),
 			Override(new(*market.FundMgr), market.NewFundMgr),
+			Override(SettlePaymentChannelsKey, settler.SettlePaymentChannels),
 		),
 
 		// miner

paychmgr/settler/settler.go

@@ -0,0 +1,114 @@
+package settler
+
+import (
+	"context"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/events"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/node/impl/full"
+	payapi "github.com/filecoin-project/lotus/node/impl/paych"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/builtin"
+	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
+	"go.uber.org/fx"
+)
+
+// API are the dependencies need to run the payment channel settler
+type API struct {
+	fx.In
+
+	full.ChainAPI
+	full.StateAPI
+	payapi.PaychAPI
+}
+
+type paymentChannelSettler struct {
+	ctx   context.Context
+	paych payapi.PaychAPI
+}
+
+type eventAPI struct {
+	full.ChainAPI
+	full.StateAPI
+}
+
+// SettlePaymentChannels checks the chain for events related to payment channels settling and
+// submits any vouchers for inbound channels tracked for this node
+func SettlePaymentChannels(lc fx.Lifecycle, api API) error {
+	lc.Append(fx.Hook{
+		OnStart: func(ctx context.Context) error {
+			pcs := &paymentChannelSettler{
+				ctx:   ctx,
+				paych: api.PaychAPI,
+			}
+
+			ev := events.NewEvents(ctx, &eventAPI{api.ChainAPI, api.StateAPI})
+			return ev.Called(pcs.check, pcs.messageHandler, pcs.revertHandler, int(build.MessageConfidence+1), events.NoTimeout, pcs.matcher)
+		},
+	})
+	return nil
+}
+
+func (pcs *paymentChannelSettler) check(ts *types.TipSet) (done bool, more bool, err error) {
+	return false, true, nil
+}
+
+func (pcs *paymentChannelSettler) messageHandler(msg *types.Message, rec *types.MessageReceipt, ts *types.TipSet, curH abi.ChainEpoch) (more bool, err error) {
+	vouchers, err := pcs.paych.PaychVoucherList(pcs.ctx, msg.To)
+	if err != nil {
+		return true, err
+	}
+
+	bestByLane := make(map[uint64]*paych.SignedVoucher)
+	for _, voucher := range vouchers {
+		spendable, err := pcs.paych.PaychVoucherCheckSpendable(pcs.ctx, msg.To, voucher, nil, nil)
+		if err != nil {
+			return true, err
+		}
+		if spendable {
+			if bestByLane[voucher.Lane] == nil || voucher.Amount.GreaterThan(bestByLane[voucher.Lane].Amount) {
+				bestByLane[voucher.Lane] = voucher
+			}
+		}
+	}
+	for _, voucher := range bestByLane {
+		_, err := pcs.paych.PaychVoucherSubmit(pcs.ctx, msg.To, voucher)
+		if err != nil {
+			return true, err
+		}
+		// TODO: StateWaitMsg?
+	}
+	return true, nil
+}
+
+func (pcs *paymentChannelSettler) revertHandler(ctx context.Context, ts *types.TipSet) error {
+	// TODO: fill in
+	return nil
+}
+
+func (pcs *paymentChannelSettler) matcher(msg *types.Message) (matchOnce bool, matched bool, err error) {
+	// Check if this is a settle payment channel message
+	if msg.Method != builtin.MethodsPaych.Settle {
+		return false, false, nil
+	}
+	// Check if this payment channel is of concern to this miner (i.e. tracked in payment channel store),
+	// and its inbound (i.e. we're getting vouchers that we may need to redeem)
+	trackedAddresses, err := pcs.paych.PaychList(pcs.ctx)
+	if err != nil {
+		return false, false, err
+	}
+	for _, addr := range trackedAddresses {
+		if msg.To == addr {
+			status, err := pcs.paych.PaychStatus(pcs.ctx, addr)
+			if err != nil {
+				return false, false, err
+			}
+			if status.Direction == api.PCHInbound {
+				return false, true, nil
+			}
+		}
+	}
+	return false, false, nil
+}