lotus/testplans/lotus-soup/paych/stress.go

package paych

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/ipfs/go-cid"

	"github.com/filecoin-project/lotus/build"
	"github.com/filecoin-project/specs-actors/actors/builtin/paych"

	"github.com/filecoin-project/go-address"
	"github.com/filecoin-project/go-state-types/big"
	"github.com/testground/sdk-go/sync"

	"github.com/filecoin-project/lotus/testplans/lotus-soup/testkit"
)

var SendersDoneState = sync.State("senders-done")
var ReceiverReadyState = sync.State("receiver-ready")
var ReceiverAddedVouchersState = sync.State("receiver-added-vouchers")

var VoucherTopic = sync.NewTopic("voucher", &paych.SignedVoucher{})
var SettleTopic = sync.NewTopic("settle", cid.Cid{})

type ClientMode uint64

const (
	ModeSender ClientMode = iota
	ModeReceiver
)

func (cm ClientMode) String() string {
	return [...]string{"Sender", "Receiver"}[cm]
}

func getClientMode(groupSeq int64) ClientMode {
	if groupSeq == 1 {
		return ModeReceiver
	}
	return ModeSender
}

// TODO Stress is currently WIP. We found blockers in Lotus that prevent us from
//  making progress. See https://github.com/filecoin-project/lotus/issues/2297.
func Stress(t *testkit.TestEnvironment) error {
	// Dispatch/forward non-client roles to defaults.
	if t.Role != "client" {
		return testkit.HandleDefaultRole(t)
	}

	// This is a client role.
	t.RecordMessage("running payments client")

	ctx := context.Background()
	cl, err := testkit.PrepareClient(t)
	if err != nil {
		return err
	}

	// are we the receiver or a sender?
	mode := getClientMode(t.GroupSeq)
	t.RecordMessage("acting as %s", mode)

	var clients []*testkit.ClientAddressesMsg
	sctx, cancel := context.WithCancel(ctx)
	clientsCh := make(chan *testkit.ClientAddressesMsg)
	t.SyncClient.MustSubscribe(sctx, testkit.ClientsAddrsTopic, clientsCh)
	for i := 0; i < t.TestGroupInstanceCount; i++ {
		clients = append(clients, <-clientsCh)
	}
	cancel()

	switch mode {
	case ModeReceiver:
		err := runReceiver(t, ctx, cl)
		if err != nil {
			return err
		}

	case ModeSender:
		err := runSender(ctx, t, clients, cl)
		if err != nil {
			return err
		}
	}

	// Signal that the client is done
	t.SyncClient.MustSignalEntry(ctx, testkit.StateDone)

	// Signal to the miners to stop mining
	t.SyncClient.MustSignalEntry(ctx, testkit.StateStopMining)

	return nil
}

func runSender(ctx context.Context, t *testkit.TestEnvironment, clients []*testkit.ClientAddressesMsg, cl *testkit.LotusClient) error {
	var (
		// lanes to open; vouchers will be distributed across these lanes in round-robin fashion
		laneCount = t.IntParam("lane_count")
		// number of vouchers to send on each lane
		vouchersPerLane = t.IntParam("vouchers_per_lane")
		// increments in which to send payment vouchers
		increments = big.Mul(big.NewInt(int64(t.IntParam("increments"))), big.NewInt(int64(build.FilecoinPrecision)))
		// channel amount should be enough to cover all vouchers
		channelAmt = big.Mul(big.NewInt(int64(laneCount*vouchersPerLane)), increments)
	)

	// Lock up funds in the payment channel.
	recv := findReceiver(clients)
	balance, err := cl.FullApi.WalletBalance(ctx, cl.Wallet.Address)
	if err != nil {
		return fmt.Errorf("failed to acquire wallet balance: %w", err)
	}

	t.RecordMessage("my balance: %d", balance)
	t.RecordMessage("creating payment channel; from=%s, to=%s, funds=%d", cl.Wallet.Address, recv.WalletAddr, channelAmt)

	pid := os.Getpid()
	t.RecordMessage("sender pid: %d", pid)

	time.Sleep(20 * time.Second)

	channel, err := cl.FullApi.PaychGet(ctx, cl.Wallet.Address, recv.WalletAddr, channelAmt)
	if err != nil {
		return fmt.Errorf("failed to create payment channel: %w", err)
	}

	if addr := channel.Channel; addr != address.Undef {
		return fmt.Errorf("expected an Undef channel address, got: %s", addr)
	}

	t.RecordMessage("payment channel created; msg_cid=%s", channel.WaitSentinel)
	t.RecordMessage("waiting for payment channel message to appear on chain")

	// wait for the channel creation message to appear on chain.
	_, err = cl.FullApi.StateWaitMsg(ctx, channel.WaitSentinel, 2)
	if err != nil {
		return fmt.Errorf("failed while waiting for payment channel creation msg to appear on chain: %w", err)
	}

	// need to wait so that the channel is tracked.
	// the full API waits for build.MessageConfidence (=1 in tests) before tracking the channel.
	// we wait for 2 confirmations, so we have the assurance the channel is tracked.

	t.RecordMessage("get payment channel address")
	channelAddr, err := cl.FullApi.PaychGetWaitReady(ctx, channel.WaitSentinel)
	if err != nil {
		return fmt.Errorf("failed to get payment channel address: %w", err)
	}

	t.RecordMessage("channel address: %s", channelAddr)
	t.RecordMessage("allocating lanes; count=%d", laneCount)

	// allocate as many lanes as required
	var lanes []uint64
	for i := 0; i < laneCount; i++ {
		lane, err := cl.FullApi.PaychAllocateLane(ctx, channelAddr)
		if err != nil {
			return fmt.Errorf("failed to allocate lane: %w", err)
		}
		lanes = append(lanes, lane)
	}

	t.RecordMessage("lanes allocated; count=%d", laneCount)

	<-t.SyncClient.MustBarrier(ctx, ReceiverReadyState, 1).C

	t.RecordMessage("sending payments in round-robin fashion across lanes; increments=%d", increments)

	// create vouchers
	remaining := channelAmt
	for i := 0; i < vouchersPerLane; i++ {
		for _, lane := range lanes {
			voucherAmt := big.Mul(big.NewInt(int64(i+1)), increments)
			voucher, err := cl.FullApi.PaychVoucherCreate(ctx, channelAddr, voucherAmt, lane)
			if err != nil {
				return fmt.Errorf("failed to create voucher: %w", err)
			}
			t.RecordMessage("payment voucher created; lane=%d, nonce=%d, amount=%d", voucher.Voucher.Lane, voucher.Voucher.Nonce, voucher.Voucher.Amount)

			_, err = t.SyncClient.Publish(ctx, VoucherTopic, voucher.Voucher)
			if err != nil {
				return fmt.Errorf("failed to publish voucher: %w", err)
			}

			remaining = big.Sub(remaining, increments)
			t.RecordMessage("remaining balance: %d", remaining)
		}
	}

	t.RecordMessage("finished sending all payment vouchers")

	// Inform the receiver that all vouchers have been created
	t.SyncClient.MustSignalEntry(ctx, SendersDoneState)

	// Wait for the receiver to add all vouchers
	<-t.SyncClient.MustBarrier(ctx, ReceiverAddedVouchersState, 1).C

	t.RecordMessage("settle channel")

	// Settle the channel. When the receiver sees the settle message, they
	// should automatically submit all vouchers.
	settleMsgCid, err := cl.FullApi.PaychSettle(ctx, channelAddr)
	if err != nil {
		return fmt.Errorf("failed to settle payment channel: %w", err)
	}

	t.SyncClient.Publish(ctx, SettleTopic, settleMsgCid)
	if err != nil {
		return fmt.Errorf("failed to publish settle message cid: %w", err)
	}

	return nil
}

func findReceiver(clients []*testkit.ClientAddressesMsg) *testkit.ClientAddressesMsg {
	for _, c := range clients {
		if getClientMode(c.GroupSeq) == ModeReceiver {
			return c
		}
	}
	return nil
}

func runReceiver(t *testkit.TestEnvironment, ctx context.Context, cl *testkit.LotusClient) error {
	// lanes to open; vouchers will be distributed across these lanes in round-robin fashion
	laneCount := t.IntParam("lane_count")
	// number of vouchers to send on each lane
	vouchersPerLane := t.IntParam("vouchers_per_lane")
	totalVouchers := laneCount * vouchersPerLane

	vouchers := make(chan *paych.SignedVoucher)
	vouchersSub, err := t.SyncClient.Subscribe(ctx, VoucherTopic, vouchers)
	if err != nil {
		return fmt.Errorf("failed to subscribe to voucher topic: %w", err)
	}

	settleMsgChan := make(chan cid.Cid)
	settleSub, err := t.SyncClient.Subscribe(ctx, SettleTopic, settleMsgChan)
	if err != nil {
		return fmt.Errorf("failed to subscribe to settle topic: %w", err)
	}

	// inform the clients that the receiver is ready for incoming vouchers
	t.SyncClient.MustSignalEntry(ctx, ReceiverReadyState)

	t.RecordMessage("adding %d payment vouchers", totalVouchers)

	// Add each of the vouchers
	var addedVouchers []*paych.SignedVoucher
	for i := 0; i < totalVouchers; i++ {
		v := <-vouchers
		addedVouchers = append(addedVouchers, v)

		_, err := cl.FullApi.PaychVoucherAdd(ctx, v.ChannelAddr, v, nil, big.NewInt(0))
		if err != nil {
			return fmt.Errorf("failed to add voucher: %w", err)
		}
		spendable, err := cl.FullApi.PaychVoucherCheckSpendable(ctx, v.ChannelAddr, v, nil, nil)
		if err != nil {
			return fmt.Errorf("failed to check voucher spendable: %w", err)
		}
		if !spendable {
			return fmt.Errorf("expected voucher %d to be spendable", i)
		}

		t.RecordMessage("payment voucher added; lane=%d, nonce=%d, amount=%d", v.Lane, v.Nonce, v.Amount)
	}

	vouchersSub.Done()

	t.RecordMessage("finished adding all payment vouchers")

	// Inform the clients that the receiver has added all vouchers
	t.SyncClient.MustSignalEntry(ctx, ReceiverAddedVouchersState)

	// Wait for the settle message (put on chain by the sender)
	t.RecordMessage("waiting for client to put settle message on chain")
	settleMsgCid := <-settleMsgChan
	settleSub.Done()

	time.Sleep(5 * time.Second)

	t.RecordMessage("waiting for confirmation of settle message on chain: %s", settleMsgCid)
	_, err = cl.FullApi.StateWaitMsg(ctx, settleMsgCid, 10)
	if err != nil {
		return fmt.Errorf("failed to wait for settle message: %w", err)
	}

	// Note: Once the receiver sees the settle message on chain, it will
	// automatically call submit voucher with the best vouchers

	// TODO: Uncomment this section once this PR is merged:
	// https://github.com/filecoin-project/lotus/pull/3197
	//t.RecordMessage("checking that all %d vouchers are no longer spendable", len(addedVouchers))
	//for i, v := range addedVouchers {
	//	spendable, err := cl.FullApi.PaychVoucherCheckSpendable(ctx, v.ChannelAddr, v, nil, nil)
	//	if err != nil {
	//		return fmt.Errorf("failed to check voucher spendable: %w", err)
	//	}
	//	// Should no longer be spendable because the best voucher has been submitted
	//	if spendable {
	//		return fmt.Errorf("expected voucher %d to no longer be spendable", i)
	//	}
	//}

	t.RecordMessage("all vouchers were submitted successfully")

	return nil
}