Merge remote-tracking branch 'origin/master' into feat/async-restartable-workers

cmd/lotus-storage-miner/actor.go

@@ -40,6 +40,8 @@ var actorCmd = &cli.Command{
 		actorSetPeeridCmd,
 		actorSetOwnerCmd,
 		actorControl,
+		actorProposeChangeWorker,
+		actorConfirmChangeWorker,
 	},
 }
 
@@ -698,3 +700,221 @@ var actorSetOwnerCmd = &cli.Command{
 		return nil
 	},
 }
+
+var actorProposeChangeWorker = &cli.Command{
+	Name:      "propose-change-worker",
+	Usage:     "Propose a worker address change",
+	ArgsUsage: "[address]",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "really-do-it",
+			Usage: "Actually send transaction performing the action",
+			Value: false,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		if !cctx.Args().Present() {
+			return fmt.Errorf("must pass address of new worker address")
+		}
+
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		api, acloser, err := lcli.GetFullNodeAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer acloser()
+
+		ctx := lcli.ReqContext(cctx)
+
+		na, err := address.NewFromString(cctx.Args().First())
+		if err != nil {
+			return err
+		}
+
+		newAddr, err := api.StateLookupID(ctx, na, types.EmptyTSK)
+		if err != nil {
+			return err
+		}
+
+		maddr, err := nodeApi.ActorAddress(ctx)
+		if err != nil {
+			return err
+		}
+
+		mi, err := api.StateMinerInfo(ctx, maddr, types.EmptyTSK)
+		if err != nil {
+			return err
+		}
+
+		if mi.NewWorker.Empty() {
+			if mi.Worker == newAddr {
+				return fmt.Errorf("worker address already set to %s", na)
+			}
+		} else {
+			if mi.NewWorker == newAddr {
+				return fmt.Errorf("change to worker address %s already pending", na)
+			}
+		}
+
+		if !cctx.Bool("really-do-it") {
+			fmt.Fprintln(cctx.App.Writer, "Pass --really-do-it to actually execute this action")
+			return nil
+		}
+
+		cwp := &miner2.ChangeWorkerAddressParams{
+			NewWorker:       newAddr,
+			NewControlAddrs: mi.ControlAddresses,
+		}
+
+		sp, err := actors.SerializeParams(cwp)
+		if err != nil {
+			return xerrors.Errorf("serializing params: %w", err)
+		}
+
+		smsg, err := api.MpoolPushMessage(ctx, &types.Message{
+			From:   mi.Owner,
+			To:     maddr,
+			Method: miner.Methods.ChangeWorkerAddress,
+			Value:  big.Zero(),
+			Params: sp,
+		}, nil)
+		if err != nil {
+			return xerrors.Errorf("mpool push: %w", err)
+		}
+
+		fmt.Fprintln(cctx.App.Writer, "Propose Message CID:", smsg.Cid())
+
+		// wait for it to get mined into a block
+		wait, err := api.StateWaitMsg(ctx, smsg.Cid(), build.MessageConfidence)
+		if err != nil {
+			return err
+		}
+
+		// check it executed successfully
+		if wait.Receipt.ExitCode != 0 {
+			fmt.Fprintln(cctx.App.Writer, "Propose worker change failed!")
+			return err
+		}
+
+		mi, err = api.StateMinerInfo(ctx, maddr, wait.TipSet)
+		if err != nil {
+			return err
+		}
+		if mi.NewWorker != newAddr {
+			return fmt.Errorf("Proposed worker address change not reflected on chain: expected '%s', found '%s'", na, mi.NewWorker)
+		}
+
+		fmt.Fprintf(cctx.App.Writer, "Worker key change to %s successfully proposed.\n", na)
+		fmt.Fprintf(cctx.App.Writer, "Call 'confirm-change-worker' at or after height %d to complete.\n", mi.WorkerChangeEpoch)
+
+		return nil
+	},
+}
+
+var actorConfirmChangeWorker = &cli.Command{
+	Name:      "confirm-change-worker",
+	Usage:     "Confirm a worker address change",
+	ArgsUsage: "[address]",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "really-do-it",
+			Usage: "Actually send transaction performing the action",
+			Value: false,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		if !cctx.Args().Present() {
+			return fmt.Errorf("must pass address of new worker address")
+		}
+
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		api, acloser, err := lcli.GetFullNodeAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer acloser()
+
+		ctx := lcli.ReqContext(cctx)
+
+		na, err := address.NewFromString(cctx.Args().First())
+		if err != nil {
+			return err
+		}
+
+		newAddr, err := api.StateLookupID(ctx, na, types.EmptyTSK)
+		if err != nil {
+			return err
+		}
+
+		maddr, err := nodeApi.ActorAddress(ctx)
+		if err != nil {
+			return err
+		}
+
+		mi, err := api.StateMinerInfo(ctx, maddr, types.EmptyTSK)
+		if err != nil {
+			return err
+		}
+
+		if mi.NewWorker.Empty() {
+			return xerrors.Errorf("no worker key change proposed")
+		} else if mi.NewWorker != newAddr {
+			return xerrors.Errorf("worker key %s does not match current worker key proposal %s", newAddr, mi.NewWorker)
+		}
+
+		if head, err := api.ChainHead(ctx); err != nil {
+			return xerrors.Errorf("failed to get the chain head: %w", err)
+		} else if head.Height() < mi.WorkerChangeEpoch {
+			return xerrors.Errorf("worker key change cannot be confirmed until %d, current height is %d", mi.WorkerChangeEpoch, head.Height())
+		}
+
+		if !cctx.Bool("really-do-it") {
+			fmt.Fprintln(cctx.App.Writer, "Pass --really-do-it to actually execute this action")
+			return nil
+		}
+
+		smsg, err := api.MpoolPushMessage(ctx, &types.Message{
+			From:   mi.Owner,
+			To:     maddr,
+			Method: miner.Methods.ConfirmUpdateWorkerKey,
+			Value:  big.Zero(),
+		}, nil)
+		if err != nil {
+			return xerrors.Errorf("mpool push: %w", err)
+		}
+
+		fmt.Fprintln(cctx.App.Writer, "Confirm Message CID:", smsg.Cid())
+
+		// wait for it to get mined into a block
+		wait, err := api.StateWaitMsg(ctx, smsg.Cid(), build.MessageConfidence)
+		if err != nil {
+			return err
+		}
+
+		// check it executed successfully
+		if wait.Receipt.ExitCode != 0 {
+			fmt.Fprintln(cctx.App.Writer, "Worker change failed!")
+			return err
+		}
+
+		mi, err = api.StateMinerInfo(ctx, maddr, wait.TipSet)
+		if err != nil {
+			return err
+		}
+		if mi.Worker != newAddr {
+			return fmt.Errorf("Confirmed worker address change not reflected on chain: expected '%s', found '%s'", newAddr, mi.Worker)
+		}
+
+		return nil
+	},
+}

cmd/lotus-storage-miner/actor_test.go

@@ -0,0 +1,164 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"flag"
+	"fmt"
+	"regexp"
+	"strconv"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/stretchr/testify/require"
+	"github.com/urfave/cli/v2"
+
+	"github.com/filecoin-project/go-state-types/abi"
+
+	"github.com/filecoin-project/lotus/api/test"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/actors/policy"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/lib/lotuslog"
+	"github.com/filecoin-project/lotus/node/repo"
+	builder "github.com/filecoin-project/lotus/node/test"
+)
+
+func TestWorkerKeyChange(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping test in short mode")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	_ = logging.SetLogLevel("*", "INFO")
+
+	policy.SetConsensusMinerMinPower(abi.NewStoragePower(2048))
+	policy.SetSupportedProofTypes(abi.RegisteredSealProof_StackedDrg2KiBV1)
+	policy.SetMinVerifiedDealSize(abi.NewStoragePower(256))
+
+	lotuslog.SetupLogLevels()
+	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("chainstore", "ERROR")
+	logging.SetLogLevel("chain", "ERROR")
+	logging.SetLogLevel("pubsub", "ERROR")
+	logging.SetLogLevel("sub", "ERROR")
+	logging.SetLogLevel("storageminer", "ERROR")
+
+	blocktime := 1 * time.Millisecond
+
+	n, sn := builder.MockSbBuilder(t, []test.FullNodeOpts{test.FullNodeWithUpgradeAt(1), test.FullNodeWithUpgradeAt(1)}, test.OneMiner)
+
+	client1 := n[0]
+	client2 := n[1]
+
+	// Connect the nodes.
+	addrinfo, err := client1.NetAddrsListen(ctx)
+	require.NoError(t, err)
+	err = client2.NetConnect(ctx, addrinfo)
+	require.NoError(t, err)
+
+	output := bytes.NewBuffer(nil)
+	run := func(cmd *cli.Command, args ...string) error {
+		app := cli.NewApp()
+		app.Metadata = map[string]interface{}{
+			"repoType":         repo.StorageMiner,
+			"testnode-full":    n[0],
+			"testnode-storage": sn[0],
+		}
+		app.Writer = output
+		build.RunningNodeType = build.NodeMiner
+
+		fs := flag.NewFlagSet("", flag.ContinueOnError)
+		for _, f := range cmd.Flags {
+			if err := f.Apply(fs); err != nil {
+				return err
+			}
+		}
+		require.NoError(t, fs.Parse(args))
+
+		cctx := cli.NewContext(app, fs, nil)
+		return cmd.Action(cctx)
+	}
+
+	// setup miner
+	mine := int64(1)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for atomic.LoadInt64(&mine) == 1 {
+			time.Sleep(blocktime)
+			if err := sn[0].MineOne(ctx, test.MineNext); err != nil {
+				t.Error(err)
+			}
+		}
+	}()
+	defer func() {
+		atomic.AddInt64(&mine, -1)
+		fmt.Println("shutting down mining")
+		<-done
+	}()
+
+	newKey, err := client1.WalletNew(ctx, types.KTBLS)
+	require.NoError(t, err)
+
+	// Initialize wallet.
+	test.SendFunds(ctx, t, client1, newKey, abi.NewTokenAmount(0))
+
+	require.NoError(t, run(actorProposeChangeWorker, "--really-do-it", newKey.String()))
+
+	result := output.String()
+	output.Reset()
+
+	require.Contains(t, result, fmt.Sprintf("Worker key change to %s successfully proposed.", newKey))
+
+	epochRe := regexp.MustCompile("at or after height (?P<epoch>[0-9]+) to complete")
+	matches := epochRe.FindStringSubmatch(result)
+	require.NotNil(t, matches)
+	targetEpoch, err := strconv.Atoi(matches[1])
+	require.NoError(t, err)
+	require.NotZero(t, targetEpoch)
+
+	// Too early.
+	require.Error(t, run(actorConfirmChangeWorker, "--really-do-it", newKey.String()))
+	output.Reset()
+
+	for {
+		head, err := client1.ChainHead(ctx)
+		require.NoError(t, err)
+		if head.Height() >= abi.ChainEpoch(targetEpoch) {
+			break
+		}
+		build.Clock.Sleep(10 * blocktime)
+	}
+	require.NoError(t, run(actorConfirmChangeWorker, "--really-do-it", newKey.String()))
+	output.Reset()
+
+	head, err := client1.ChainHead(ctx)
+	require.NoError(t, err)
+
+	// Wait for finality (worker key switch).
+	targetHeight := head.Height() + policy.ChainFinality
+	for {
+		head, err := client1.ChainHead(ctx)
+		require.NoError(t, err)
+		if head.Height() >= targetHeight {
+			break
+		}
+		build.Clock.Sleep(10 * blocktime)
+	}
+
+	// Make sure the other node can catch up.
+	for i := 0; i < 20; i++ {
+		head, err := client2.ChainHead(ctx)
+		require.NoError(t, err)
+		if head.Height() >= targetHeight {
+			return
+		}
+		build.Clock.Sleep(10 * blocktime)
+	}
+	t.Fatal("failed to reach target epoch on the second miner")
+}

cmd/lotus-storage-miner/market.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -13,8 +14,10 @@ import (
 
 	tm "github.com/buger/goterm"
 	"github.com/docker/go-units"
+	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-cidutil/cidenc"
+	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/multiformats/go-multibase"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
@@ -569,6 +572,128 @@ var dataTransfersCmd = &cli.Command{
 	Usage: "Manage data transfers",
 	Subcommands: []*cli.Command{
 		transfersListCmd,
+		marketRestartTransfer,
+		marketCancelTransfer,
+	},
+}
+
+var marketRestartTransfer = &cli.Command{
+	Name:  "restart",
+	Usage: "Force restart a stalled data transfer",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:  "peerid",
+			Usage: "narrow to transfer with specific peer",
+		},
+		&cli.BoolFlag{
+			Name:  "initiator",
+			Usage: "specify only transfers where peer is/is not initiator",
+			Value: false,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		if !cctx.Args().Present() {
+			return cli.ShowCommandHelp(cctx, cctx.Command.Name)
+		}
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := lcli.ReqContext(cctx)
+
+		transferUint, err := strconv.ParseUint(cctx.Args().First(), 10, 64)
+		if err != nil {
+			return fmt.Errorf("Error reading transfer ID: %w", err)
+		}
+		transferID := datatransfer.TransferID(transferUint)
+		initiator := cctx.Bool("initiator")
+		var other peer.ID
+		if pidstr := cctx.String("peerid"); pidstr != "" {
+			p, err := peer.Decode(pidstr)
+			if err != nil {
+				return err
+			}
+			other = p
+		} else {
+			channels, err := nodeApi.MarketListDataTransfers(ctx)
+			if err != nil {
+				return err
+			}
+			found := false
+			for _, channel := range channels {
+				if channel.IsInitiator == initiator && channel.TransferID == transferID {
+					other = channel.OtherPeer
+					found = true
+					break
+				}
+			}
+			if !found {
+				return errors.New("unable to find matching data transfer")
+			}
+		}
+
+		return nodeApi.MarketRestartDataTransfer(ctx, transferID, other, initiator)
+	},
+}
+
+var marketCancelTransfer = &cli.Command{
+	Name:  "cancel",
+	Usage: "Force cancel a data transfer",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:  "peerid",
+			Usage: "narrow to transfer with specific peer",
+		},
+		&cli.BoolFlag{
+			Name:  "initiator",
+			Usage: "specify only transfers where peer is/is not initiator",
+			Value: false,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		if !cctx.Args().Present() {
+			return cli.ShowCommandHelp(cctx, cctx.Command.Name)
+		}
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := lcli.ReqContext(cctx)
+
+		transferUint, err := strconv.ParseUint(cctx.Args().First(), 10, 64)
+		if err != nil {
+			return fmt.Errorf("Error reading transfer ID: %w", err)
+		}
+		transferID := datatransfer.TransferID(transferUint)
+		initiator := cctx.Bool("initiator")
+		var other peer.ID
+		if pidstr := cctx.String("peerid"); pidstr != "" {
+			p, err := peer.Decode(pidstr)
+			if err != nil {
+				return err
+			}
+			other = p
+		} else {
+			channels, err := nodeApi.MarketListDataTransfers(ctx)
+			if err != nil {
+				return err
+			}
+			found := false
+			for _, channel := range channels {
+				if channel.IsInitiator == initiator && channel.TransferID == transferID {
+					other = channel.OtherPeer
+					found = true
+					break
+				}
+			}
+			if !found {
+				return errors.New("unable to find matching data transfer")
+			}
+		}
+
+		return nodeApi.MarketCancelDataTransfer(ctx, transferID, other, initiator)
 	},
 }
 
@@ -589,6 +714,10 @@ var transfersListCmd = &cli.Command{
 			Name:  "watch",
 			Usage: "watch deal updates in real-time, rather than a one time list",
 		},
+		&cli.BoolFlag{
+			Name:  "show-failed",
+			Usage: "show failed/cancelled transfers",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		api, closer, err := lcli.GetStorageMinerAPI(cctx)
@@ -606,7 +735,7 @@ var transfersListCmd = &cli.Command{
 		completed := cctx.Bool("completed")
 		color := cctx.Bool("color")
 		watch := cctx.Bool("watch")
-
+		showFailed := cctx.Bool("show-failed")
 		if watch {
 			channelUpdates, err := api.MarketDataTransferUpdates(ctx)
 			if err != nil {
@@ -618,7 +747,7 @@ var transfersListCmd = &cli.Command{
 
 				tm.MoveCursor(1, 1)
 
-				lcli.OutputDataTransferChannels(tm.Screen, channels, completed, color)
+				lcli.OutputDataTransferChannels(tm.Screen, channels, completed, color, showFailed)
 
 				tm.Flush()
 
@@ -643,7 +772,7 @@ var transfersListCmd = &cli.Command{
 				}
 			}
 		}
-		lcli.OutputDataTransferChannels(os.Stdout, channels, completed, color)
+		lcli.OutputDataTransferChannels(os.Stdout, channels, completed, color, showFailed)
 		return nil
 	},
 }

cmd/lotus-storage-miner/sectors.go

@@ -219,7 +219,7 @@ var sectorsListCmd = &cli.Command{
 			tablewriter.Col("Deals"),
 			tablewriter.Col("DealWeight"),
 			tablewriter.NewLineCol("Error"),
-			tablewriter.NewLineCol("EarlyExpiration"))
+			tablewriter.NewLineCol("RecoveryTimeout"))
 
 		fast := cctx.Bool("fast")
 
@@ -281,7 +281,7 @@ var sectorsListCmd = &cli.Command{
 						}
 
 						if st.Early > 0 {
-							m["EarlyExpiration"] = color.YellowString(lcli.EpochTime(head.Height(), st.Early))
+							m["RecoveryTimeout"] = color.YellowString(lcli.EpochTime(head.Height(), st.Early))
 						}
 					}
 				}