feat(paych): add simple integration test

api/api_full.go

@@ -367,7 +367,8 @@ type FullNode interface {
 	PaychGet(ctx context.Context, from, to address.Address, ensureFunds types.BigInt) (*ChannelInfo, error)
 	PaychList(context.Context) ([]address.Address, error)
 	PaychStatus(context.Context, address.Address) (*PaychStatus, error)
-	PaychClose(context.Context, address.Address) (cid.Cid, error)
+	PaychSettle(context.Context, address.Address) (cid.Cid, error)
+	PaychCollect(context.Context, address.Address) (cid.Cid, error)
 	PaychAllocateLane(ctx context.Context, ch address.Address) (uint64, error)
 	PaychNewPayment(ctx context.Context, from, to address.Address, vouchers []VoucherSpec) (*PaymentInfo, error)
 	PaychVoucherCheckValid(context.Context, address.Address, *paych.SignedVoucher) error

api/apistruct/struct.go

@@ -180,7 +180,8 @@ type FullNodeStruct struct {
 		PaychGet                   func(ctx context.Context, from, to address.Address, ensureFunds types.BigInt) (*api.ChannelInfo, error)   `perm:"sign"`
 		PaychList                  func(context.Context) ([]address.Address, error)                                                          `perm:"read"`
 		PaychStatus                func(context.Context, address.Address) (*api.PaychStatus, error)                                          `perm:"read"`
-		PaychClose                 func(context.Context, address.Address) (cid.Cid, error)                                                   `perm:"sign"`
+		PaychSettle                func(context.Context, address.Address) (cid.Cid, error)                                                   `perm:"sign"`
+		PaychCollect               func(context.Context, address.Address) (cid.Cid, error)                                                   `perm:"sign"`
 		PaychAllocateLane          func(context.Context, address.Address) (uint64, error)                                                    `perm:"sign"`
 		PaychNewPayment            func(ctx context.Context, from, to address.Address, vouchers []api.VoucherSpec) (*api.PaymentInfo, error) `perm:"sign"`
 		PaychVoucherCheck          func(context.Context, *paych.SignedVoucher) error                                                         `perm:"read"`
@@ -804,8 +805,12 @@ func (c *FullNodeStruct) PaychVoucherList(ctx context.Context, pch address.Addre
 	return c.Internal.PaychVoucherList(ctx, pch)
 }
 
-func (c *FullNodeStruct) PaychClose(ctx context.Context, a address.Address) (cid.Cid, error) {
+func (c *FullNodeStruct) PaychSettle(ctx context.Context, a address.Address) (cid.Cid, error) {
-	return c.Internal.PaychClose(ctx, a)
+	return c.Internal.PaychSettle(ctx, a)
+}
+
+func (c *FullNodeStruct) PaychCollect(ctx context.Context, a address.Address) (cid.Cid, error) {
+	return c.Internal.PaychCollect(ctx, a)
 }
 
 func (c *FullNodeStruct) PaychAllocateLane(ctx context.Context, ch address.Address) (uint64, error) {

api/test/paych.go

@@ -0,0 +1,261 @@
+package test
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/events"
+	"github.com/filecoin-project/lotus/chain/events/state"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/chain/wallet"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
+	initactor "github.com/filecoin-project/specs-actors/actors/builtin/init"
+	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
+)
+
+func TestPaymentChannels(t *testing.T, b APIBuilder, blocktime time.Duration) {
+	_ = os.Setenv("BELLMAN_NO_GPU", "1")
+
+	ctx := context.Background()
+	n, sn := b(t, 2, oneMiner)
+
+	paymentCreator := n[0]
+	paymentReceiver := n[1]
+	miner := sn[0]
+
+	// get everyone connected
+	addrs, err := paymentCreator.NetAddrsListen(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := paymentReceiver.NetConnect(ctx, addrs); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := miner.NetConnect(ctx, addrs); err != nil {
+		t.Fatal(err)
+	}
+
+	// start mining blocks
+	bm := newBlockMiner(ctx, t, miner, blocktime)
+	bm.mineBlocks()
+
+	// send some funds to register the receiver
+	receiverAddr, err := paymentReceiver.WalletNew(ctx, wallet.ActSigType("secp256k1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sendFunds(ctx, t, paymentCreator, receiverAddr, abi.NewTokenAmount(1000))
+
+	// setup the payment channel
+	createrAddr, err := paymentCreator.WalletDefaultAddress(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	initBalance, err := paymentCreator.WalletBalance(ctx, createrAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	channelInfo, err := paymentCreator.PaychGet(ctx, createrAddr, receiverAddr, abi.NewTokenAmount(100000))
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err := paymentCreator.StateWaitMsg(ctx, channelInfo.ChannelMessage, 1)
+	if res.Receipt.ExitCode != 0 {
+		t.Fatal("did not successfully create payment channel")
+	}
+	var params initactor.ExecReturn
+	err = params.UnmarshalCBOR(bytes.NewReader(res.Receipt.Return))
+	if err != nil {
+		t.Fatal(err)
+	}
+	channel := params.RobustAddress
+	// allocate three lanes
+	var lanes []uint64
+	for i := 0; i < 3; i++ {
+		lane, err := paymentCreator.PaychAllocateLane(ctx, channel)
+		if err != nil {
+			t.Fatal(err)
+		}
+		lanes = append(lanes, lane)
+	}
+
+	// make two vouchers each for each lane, then save on the other side
+	for _, lane := range lanes {
+		vouch1, err := paymentCreator.PaychVoucherCreate(ctx, channel, abi.NewTokenAmount(1000), lane)
+		if err != nil {
+			t.Fatal(err)
+		}
+		vouch2, err := paymentCreator.PaychVoucherCreate(ctx, channel, abi.NewTokenAmount(2000), lane)
+		if err != nil {
+			t.Fatal(err)
+		}
+		delta1, err := paymentReceiver.PaychVoucherAdd(ctx, channel, vouch1, nil, abi.NewTokenAmount(1000))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !delta1.Equals(abi.NewTokenAmount(1000)) {
+			t.Fatal("voucher didn't have the right amount")
+		}
+		delta2, err := paymentReceiver.PaychVoucherAdd(ctx, channel, vouch2, nil, abi.NewTokenAmount(1000))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !delta2.Equals(abi.NewTokenAmount(1000)) {
+			t.Fatal("voucher didn't have the right amount")
+		}
+	}
+
+	// settle the payment channel
+	settleMsgCid, err := paymentCreator.PaychSettle(ctx, channel)
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err = paymentCreator.StateWaitMsg(ctx, settleMsgCid, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Receipt.ExitCode != 0 {
+		t.Fatal("Unable to settle payment channel")
+	}
+
+	// wait for the receiver to submit their vouchers
+	ev := events.NewEvents(ctx, paymentCreator)
+	preds := state.NewStatePredicates(paymentCreator)
+	finished := make(chan struct{})
+	err = ev.StateChanged(func(ts *types.TipSet) (done bool, more bool, err error) {
+		act, err := paymentCreator.StateReadState(ctx, channel, ts.Key())
+		if err != nil {
+			return false, false, err
+		}
+		state := act.State.(paych.State)
+		if state.ToSend.GreaterThanEqual(abi.NewTokenAmount(6000)) {
+			return true, false, nil
+		}
+		return false, true, nil
+	}, func(oldTs, newTs *types.TipSet, states events.StateChange, curH abi.ChainEpoch) (more bool, err error) {
+		toSendChange := states.(*state.PayChToSendChange)
+		if toSendChange.NewToSend.GreaterThanEqual(abi.NewTokenAmount(6000)) {
+			close(finished)
+			return false, nil
+		}
+		return true, nil
+	}, func(ctx context.Context, ts *types.TipSet) error {
+		return nil
+	}, int(build.MessageConfidence)+1, build.SealRandomnessLookbackLimit, func(oldTs, newTs *types.TipSet) (bool, events.StateChange, error) {
+		return preds.OnPaymentChannelActorChanged(channel, preds.OnToSendAmountChanges())(ctx, oldTs.Key(), newTs.Key())
+	})
+
+	<-finished
+
+	// collect funds (from receiver, though either party can do it)
+	collectMsg, err := paymentReceiver.PaychCollect(ctx, channel)
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err = paymentReceiver.StateWaitMsg(ctx, collectMsg, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Receipt.ExitCode != 0 {
+		t.Fatal("unable to collect on payment channel")
+	}
+
+	// Finally, check the balance for the receiver and creator
+	currentCreatorBalance, err := paymentCreator.WalletBalance(ctx, createrAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !big.Sub(initBalance, currentCreatorBalance).Equals(abi.NewTokenAmount(7000)) {
+		t.Fatal("did not send correct funds from creator")
+	}
+	currentReceiverBalance, err := paymentReceiver.WalletBalance(ctx, receiverAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !currentReceiverBalance.Equals(abi.NewTokenAmount(7000)) {
+		t.Fatal("did not receive correct funds on receiver")
+	}
+
+	// shut down mining
+	bm.stop()
+}
+
+type blockMiner struct {
+	ctx       context.Context
+	t         *testing.T
+	miner     TestStorageNode
+	blocktime time.Duration
+	mine      int64
+	done      chan struct{}
+}
+
+func newBlockMiner(ctx context.Context, t *testing.T, miner TestStorageNode, blocktime time.Duration) *blockMiner {
+	return &blockMiner{
+		ctx:       ctx,
+		t:         t,
+		miner:     miner,
+		blocktime: blocktime,
+		mine:      int64(1),
+		done:      make(chan struct{}),
+	}
+}
+
+func (bm *blockMiner) mineBlocks() {
+	time.Sleep(time.Second)
+	go func() {
+		defer close(bm.done)
+		for atomic.LoadInt64(&bm.mine) == 1 {
+			time.Sleep(bm.blocktime)
+			if err := bm.miner.MineOne(bm.ctx, func(bool, error) {}); err != nil {
+				bm.t.Error(err)
+			}
+		}
+	}()
+}
+
+func (bm *blockMiner) stop() {
+	atomic.AddInt64(&bm.mine, -1)
+	fmt.Println("shutting down mining")
+	<-bm.done
+}
+
+func sendFunds(ctx context.Context, t *testing.T, sender TestNode, addr address.Address, amount abi.TokenAmount) {
+
+	senderAddr, err := sender.WalletDefaultAddress(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	msg := &types.Message{
+		From:     senderAddr,
+		To:       addr,
+		Value:    amount,
+		GasLimit: 100_000_000,
+		GasPrice: abi.NewTokenAmount(1000),
+	}
+
+	sm, err := sender.MpoolPushMessage(ctx, msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err := sender.StateWaitMsg(ctx, sm.Cid(), 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Receipt.ExitCode != 0 {
+		t.Fatal("did not successfully send money")
+	}
+}

chain/events/state/predicates.go

@@ -3,6 +3,7 @@ package state
 import (
 	"bytes"
 	"context"
+
 	"github.com/filecoin-project/go-address"
 	"github.com/ipfs/go-cid"
 	cbor "github.com/ipfs/go-ipld-cbor"
@@ -12,6 +13,7 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/builtin"
 	"github.com/filecoin-project/specs-actors/actors/builtin/market"
 	"github.com/filecoin-project/specs-actors/actors/builtin/miner"
+	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
 	"github.com/filecoin-project/specs-actors/actors/util/adt"
 
 	"github.com/filecoin-project/lotus/api/apibstore"
@@ -493,3 +495,40 @@ func (sp *StatePredicates) OnMinerPreCommitChange() DiffMinerActorStateFunc {
 		return true, precommitChanges, nil
 	}
 }
+
+// DiffPaymentChannelStateFunc is function that compares two states for the payment channel
+type DiffPaymentChannelStateFunc func(ctx context.Context, oldState *paych.State, newState *paych.State) (changed bool, user UserData, err error)
+
+// OnPaymentChannelActorChanged calls diffPaymentChannelState when the state changes for the the payment channel actor
+func (sp *StatePredicates) OnPaymentChannelActorChanged(paychAddr address.Address, diffPaymentChannelState DiffPaymentChannelStateFunc) DiffTipSetKeyFunc {
+	return sp.OnActorStateChanged(paychAddr, func(ctx context.Context, oldActorStateHead, newActorStateHead cid.Cid) (changed bool, user UserData, err error) {
+		var oldState paych.State
+		if err := sp.cst.Get(ctx, oldActorStateHead, &oldState); err != nil {
+			return false, nil, err
+		}
+		var newState paych.State
+		if err := sp.cst.Get(ctx, newActorStateHead, &newState); err != nil {
+			return false, nil, err
+		}
+		return diffPaymentChannelState(ctx, &oldState, &newState)
+	})
+}
+
+// PayChToSendChange is a difference in the amount to send on a payment channel when the money is collected
+type PayChToSendChange struct {
+	OldToSend abi.TokenAmount
+	NewToSend abi.TokenAmount
+}
+
+// OnToSendAmountChanges monitors changes on the total amount to send from one party to the other on a payment channel
+func (sp *StatePredicates) OnToSendAmountChanges() DiffPaymentChannelStateFunc {
+	return func(ctx context.Context, oldState *paych.State, newState *paych.State) (changed bool, user UserData, err error) {
+		if oldState.ToSend.Equals(newState.ToSend) {
+			return false, nil, nil
+		}
+		return true, &PayChToSendChange{
+			OldToSend: oldState.ToSend,
+			NewToSend: newState.ToSend,
+		}, nil
+	}
+}

node/impl/paych/paych.go

@@ -107,8 +107,7 @@ func (a *PaychAPI) PaychStatus(ctx context.Context, pch address.Address) (*api.P
 	}, nil
 }
 
-func (a *PaychAPI) PaychClose(ctx context.Context, addr address.Address) (cid.Cid, error) {
+func (a *PaychAPI) PaychSettle(ctx context.Context, addr address.Address) (cid.Cid, error) {
-	panic("TODO Settle logic")
 
 	ci, err := a.PaychMgr.GetChannelInfo(addr)
 	if err != nil {
@@ -143,6 +142,41 @@ func (a *PaychAPI) PaychClose(ctx context.Context, addr address.Address) (cid.Ci
 	return smsg.Cid(), nil
 }
 
+func (a *PaychAPI) PaychCollect(ctx context.Context, addr address.Address) (cid.Cid, error) {
+
+	ci, err := a.PaychMgr.GetChannelInfo(addr)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	nonce, err := a.MpoolGetNonce(ctx, ci.Control)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	msg := &types.Message{
+		To:     addr,
+		From:   ci.Control,
+		Value:  types.NewInt(0),
+		Method: builtin.MethodsPaych.Collect,
+		Nonce:  nonce,
+
+		GasLimit: 100_000_000,
+		GasPrice: types.NewInt(0),
+	}
+
+	smsg, err := a.WalletSignMessage(ctx, ci.Control, msg)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	if _, err := a.MpoolPush(ctx, smsg); err != nil {
+		return cid.Undef, err
+	}
+
+	return smsg.Cid(), nil
+}
+
 func (a *PaychAPI) PaychVoucherCheckValid(ctx context.Context, ch address.Address, sv *paych.SignedVoucher) error {
 	return a.PaychMgr.CheckVoucherValid(ctx, ch, sv)
 }

node/node_test.go

@@ -546,3 +546,13 @@ func TestCCUpgrade(t *testing.T) {
 
 	test.TestCCUpgrade(t, mockSbBuilder, 5*time.Millisecond)
 }
+
+func TestPaymentChannels(t *testing.T) {
+	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("chainstore", "ERROR")
+	logging.SetLogLevel("chain", "ERROR")
+	logging.SetLogLevel("sub", "ERROR")
+	logging.SetLogLevel("storageminer", "ERROR")
+
+	test.TestPaymentChannels(t, mockSbBuilder, 5*time.Millisecond)
+}