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

Merge remote-tracking branch 'origin/master' into next

Browse Source
This commit is contained in:
Łukasz Magiera 2020-10-06 23:54:59 +02:00
commit 2f70a91665
35 changed files with 1744 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
api/api_full.go
View File

@ -484,6 +484,12 @@ type FullNode interface {
PaychVoucherAdd(context.Context, address.Address, *paych.SignedVoucher, []byte, types.BigInt) (types.BigInt, error)
PaychVoucherList(context.Context, address.Address) ([]*paych.SignedVoucher, error)
PaychVoucherSubmit(context.Context, address.Address, *paych.SignedVoucher, []byte, []byte) (cid.Cid, error)
// CreateBackup creates node backup onder the specified file name. The
// method requires that the lotus daemon is running with the
// LOTUS_BACKUP_BASE_PATH environment variable set to some path, and that
// the path specified when calling CreateBackup is within the base path
CreateBackup(ctx context.Context, fpath string) error
}
type FileRef struct {

6
api/api_storage.go
View File

@ -101,6 +101,12 @@ type StorageMiner interface {
PiecesListCidInfos(ctx context.Context) ([]cid.Cid, error)
PiecesGetPieceInfo(ctx context.Context, pieceCid cid.Cid) (*piecestore.PieceInfo, error)
PiecesGetCIDInfo(ctx context.Context, payloadCid cid.Cid) (*piecestore.CIDInfo, error)
// CreateBackup creates node backup onder the specified file name. The
// method requires that the lotus-miner is running with the
// LOTUS_BACKUP_BASE_PATH environment variable set to some path, and that
// the path specified when calling CreateBackup is within the base path
CreateBackup(ctx context.Context, fpath string) error
}
type SealRes struct {

12
api/apistruct/struct.go
View File

@ -243,6 +243,8 @@ type FullNodeStruct struct {
PaychVoucherCreate func(context.Context, address.Address, big.Int, uint64) (*api.VoucherCreateResult, error) `perm:"sign"`
PaychVoucherList func(context.Context, address.Address) ([]*paych.SignedVoucher, error) `perm:"write"`
PaychVoucherSubmit func(context.Context, address.Address, *paych.SignedVoucher, []byte, []byte) (cid.Cid, error) `perm:"sign"`
CreateBackup func(ctx context.Context, fpath string) error `perm:"admin"`
}
}
@ -323,6 +325,8 @@ type StorageMinerStruct struct {
PiecesListCidInfos func(ctx context.Context) ([]cid.Cid, error) `perm:"read"`
PiecesGetPieceInfo func(ctx context.Context, pieceCid cid.Cid) (*piecestore.PieceInfo, error) `perm:"read"`
PiecesGetCIDInfo func(ctx context.Context, payloadCid cid.Cid) (*piecestore.CIDInfo, error) `perm:"read"`
CreateBackup func(ctx context.Context, fpath string) error `perm:"admin"`
}
}
@ -1041,6 +1045,10 @@ func (c *FullNodeStruct) PaychVoucherSubmit(ctx context.Context, ch address.Addr
return c.Internal.PaychVoucherSubmit(ctx, ch, sv, secret, proof)
}
func (c *FullNodeStruct) CreateBackup(ctx context.Context, fpath string) error {
return c.Internal.CreateBackup(ctx, fpath)
}
// StorageMinerStruct
func (c *StorageMinerStruct) ActorAddress(ctx context.Context) (address.Address, error) {
@ -1281,6 +1289,10 @@ func (c *StorageMinerStruct) PiecesGetCIDInfo(ctx context.Context, payloadCid ci
return c.Internal.PiecesGetCIDInfo(ctx, payloadCid)
}
func (c *StorageMinerStruct) CreateBackup(ctx context.Context, fpath string) error {
return c.Internal.CreateBackup(ctx, fpath)
}
// WorkerStruct
func (w *WorkerStruct) Version(ctx context.Context) (build.Version, error) {

4
build/params_testnet.go
View File

@ -5,6 +5,7 @@
package build
import (
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/chain/actors/policy"
@ -37,6 +38,9 @@ func init() {
abi.RegisteredSealProof_StackedDrg32GiBV1,
abi.RegisteredSealProof_StackedDrg64GiBV1,
)
SetAddressNetwork(address.Mainnet)
Devnet = false
}

2
chain/exchange/client.go
View File

@ -117,7 +117,7 @@ func (c *client) doRequest(
res, err := c.sendRequestToPeer(ctx, peer, req)
if err != nil {
if !xerrors.Is(err, network.ErrNoConn) {
log.Warnf("could not connect to peer %s: %s",
log.Warnf("could not send request to peer %s: %s",
peer.String(), err)
}
continue

48
chain/wallet/wallet.go
View File

@ -81,7 +81,7 @@ func (w *Wallet) findKey(addr address.Address) (*Key, error) {
return nil, nil
}
ki, err := w.keystore.Get(KNamePrefix + addr.String())
ki, err := w.tryFind(addr)
if err != nil {
if xerrors.Is(err, types.ErrKeyInfoNotFound) {
return nil, nil
@ -96,6 +96,42 @@ func (w *Wallet) findKey(addr address.Address) (*Key, error) {
return k, nil
}
func (w *Wallet) tryFind(addr address.Address) (types.KeyInfo, error) {
ki, err := w.keystore.Get(KNamePrefix + addr.String())
if err == nil {
return ki, err
}
if !xerrors.Is(err, types.ErrKeyInfoNotFound) {
return types.KeyInfo{}, err
}
// We got an ErrKeyInfoNotFound error
// Try again, this time with the testnet prefix
aChars := []rune(addr.String())
prefixRunes := []rune(address.TestnetPrefix)
if len(prefixRunes) != 1 {
return types.KeyInfo{}, xerrors.Errorf("unexpected prefix length: %d", len(prefixRunes))
}
aChars[0] = prefixRunes[0]
ki, err = w.keystore.Get(KNamePrefix + string(aChars))
if err != nil {
return types.KeyInfo{}, err
}
// We found it with the testnet prefix
// Add this KeyInfo with the mainnet prefix address string
err = w.keystore.Put(KNamePrefix+addr.String(), ki)
if err != nil {
return types.KeyInfo{}, err
}
return ki, nil
}
func (w *Wallet) Export(addr address.Address) (*types.KeyInfo, error) {
k, err := w.findKey(addr)
if err != nil {
@ -129,6 +165,7 @@ func (w *Wallet) ListAddrs() ([]address.Address, error) {
sort.Strings(all)
seen := map[address.Address]struct{}{}
out := make([]address.Address, 0, len(all))
for _, a := range all {
if strings.HasPrefix(a, KNamePrefix) {
@ -137,10 +174,19 @@ func (w *Wallet) ListAddrs() ([]address.Address, error) {
if err != nil {
return nil, xerrors.Errorf("converting name to address: %w", err)
}
if _, ok := seen[addr]; ok {
continue // got duplicate with a different prefix
}
seen[addr] = struct{}{}
out = append(out, addr)
}
}
sort.Slice(out, func(i, j int) bool {
return out[i].String() < out[j].String()
})
return out, nil
}

125
cli/backup.go Normal file
View File

@ -0,0 +1,125 @@
package cli
import (
"context"
"fmt"
"os"
logging "github.com/ipfs/go-log/v2"
"github.com/mitchellh/go-homedir"
"github.com/urfave/cli/v2"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-jsonrpc"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/repo"
)
type BackupAPI interface {
CreateBackup(ctx context.Context, fpath string) error
}
type BackupApiFn func(ctx *cli.Context) (BackupAPI, jsonrpc.ClientCloser, error)
func BackupCmd(repoFlag string, rt repo.RepoType, getApi BackupApiFn) *cli.Command {
var offlineBackup = func(cctx *cli.Context) error {
logging.SetLogLevel("badger", "ERROR") // nolint:errcheck
repoPath := cctx.String(repoFlag)
r, err := repo.NewFS(repoPath)
if err != nil {
return err
}
ok, err := r.Exists()
if err != nil {
return err
}
if !ok {
return xerrors.Errorf("repo at '%s' is not initialized", cctx.String(repoFlag))
}
lr, err := r.LockRO(rt)
if err != nil {
return xerrors.Errorf("locking repo: %w", err)
}
defer lr.Close() // nolint:errcheck
mds, err := lr.Datastore("/metadata")
if err != nil {
return xerrors.Errorf("getting metadata datastore: %w", err)
}
bds := backupds.Wrap(mds)
fpath, err := homedir.Expand(cctx.Args().First())
if err != nil {
return xerrors.Errorf("expanding file path: %w", err)
}
out, err := os.OpenFile(fpath, os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return xerrors.Errorf("opening backup file %s: %w", fpath, err)
}
if err := bds.Backup(out); err != nil {
if cerr := out.Close(); cerr != nil {
log.Errorw("error closing backup file while handling backup error", "closeErr", cerr, "backupErr", err)
}
return xerrors.Errorf("backup error: %w", err)
}
if err := out.Close(); err != nil {
return xerrors.Errorf("closing backup file: %w", err)
}
return nil
}
var onlineBackup = func(cctx *cli.Context) error {
api, closer, err := getApi(cctx)
if err != nil {
return xerrors.Errorf("getting api: %w (if the node isn't running you can use the --offline flag)", err)
}
defer closer()
err = api.CreateBackup(ReqContext(cctx), cctx.Args().First())
if err != nil {
return err
}
fmt.Println("Success")
return nil
}
return &cli.Command{
Name: "backup",
Usage: "Create node metadata backup",
Description: `The backup command writes a copy of node metadata under the specified path
Online backups:
For security reasons, the daemon must be have LOTUS_BACKUP_BASE_PATH env var set
to a path where backup files are supposed to be saved, and the path specified in
this command must be within this base path`,
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "offline",
Usage: "create backup without the node running",
},
},
ArgsUsage: "[backup file path]",
Action: func(cctx *cli.Context) error {
if cctx.Args().Len() != 1 {
return xerrors.Errorf("expected 1 argument")
}
if cctx.Bool("offline") {
return offlineBackup(cctx)
}
return onlineBackup(cctx)
},
}
}

159
cli/chain.go
View File

@ -8,6 +8,7 @@ import (
"os"
"os/exec"
"path"
"sort"
"strconv"
"strings"
"time"
@ -27,9 +28,10 @@ import (
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/stmgr"
types "github.com/filecoin-project/lotus/chain/types"
)
@ -50,6 +52,7 @@ var chainCmd = &cli.Command{
chainExportCmd,
slashConsensusFault,
chainGasPriceCmd,
chainInspectUsage,
},
}
@ -159,7 +162,7 @@ var chainGetBlock = &cli.Command{
},
}
func apiMsgCids(in []api.Message) []cid.Cid {
func apiMsgCids(in []lapi.Message) []cid.Cid {
out := make([]cid.Cid, len(in))
for k, v := range in {
out[k] = v.Cid
@ -375,6 +378,143 @@ var chainSetHeadCmd = &cli.Command{
},
}
var chainInspectUsage = &cli.Command{
Name: "inspect-usage",
Usage: "Inspect block space usage of a given tipset",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "tipset",
Usage: "specify tipset to view block space usage of",
Value: "@head",
},
&cli.IntFlag{
Name: "length",
Usage: "length of chain to inspect block space usage for",
Value: 1,
},
&cli.IntFlag{
Name: "num-results",
Usage: "number of results to print per category",
Value: 10,
},
},
Action: func(cctx *cli.Context) error {
api, closer, err := GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := ReqContext(cctx)
ts, err := LoadTipSet(ctx, cctx, api)
if err != nil {
return err
}
cur := ts
var msgs []lapi.Message
for i := 0; i < cctx.Int("length"); i++ {
pmsgs, err := api.ChainGetParentMessages(ctx, cur.Blocks()[0].Cid())
if err != nil {
return err
}
msgs = append(msgs, pmsgs...)
next, err := api.ChainGetTipSet(ctx, cur.Parents())
if err != nil {
return err
}
cur = next
}
codeCache := make(map[address.Address]cid.Cid)
lookupActorCode := func(a address.Address) (cid.Cid, error) {
c, ok := codeCache[a]
if ok {
return c, nil
}
act, err := api.StateGetActor(ctx, a, ts.Key())
if err != nil {
return cid.Undef, err
}
codeCache[a] = act.Code
return act.Code, nil
}
bySender := make(map[string]int64)
byDest := make(map[string]int64)
byMethod := make(map[string]int64)
var sum int64
for _, m := range msgs {
bySender[m.Message.From.String()] += m.Message.GasLimit
byDest[m.Message.To.String()] += m.Message.GasLimit
sum += m.Message.GasLimit
code, err := lookupActorCode(m.Message.To)
if err != nil {
return err
}
mm := stmgr.MethodsMap[code][m.Message.Method]
byMethod[mm.Name] += m.Message.GasLimit
}
type keyGasPair struct {
Key string
Gas int64
}
mapToSortedKvs := func(m map[string]int64) []keyGasPair {
var vals []keyGasPair
for k, v := range m {
vals = append(vals, keyGasPair{
Key: k,
Gas: v,
})
}
sort.Slice(vals, func(i, j int) bool {
return vals[i].Gas > vals[j].Gas
})
return vals
}
senderVals := mapToSortedKvs(bySender)
destVals := mapToSortedKvs(byDest)
methodVals := mapToSortedKvs(byMethod)
numRes := cctx.Int("num-results")
fmt.Printf("Total Gas Limit: %d\n", sum)
fmt.Printf("By Sender:\n")
for i := 0; i < numRes && i < len(senderVals); i++ {
sv := senderVals[i]
fmt.Printf("%s\t%0.2f%%\t(%d)\n", sv.Key, (100*float64(sv.Gas))/float64(sum), sv.Gas)
}
fmt.Println()
fmt.Printf("By Receiver:\n")
for i := 0; i < numRes && i < len(destVals); i++ {
sv := destVals[i]
fmt.Printf("%s\t%0.2f%%\t(%d)\n", sv.Key, (100*float64(sv.Gas))/float64(sum), sv.Gas)
}
fmt.Println()
fmt.Printf("By Method:\n")
for i := 0; i < numRes && i < len(methodVals); i++ {
sv := methodVals[i]
fmt.Printf("%s\t%0.2f%%\t(%d)\n", sv.Key, (100*float64(sv.Gas))/float64(sum), sv.Gas)
}
return nil
},
}
var chainListCmd = &cli.Command{
Name: "list",
Usage: "View a segment of the chain",
@ -648,7 +788,7 @@ var chainGetCmd = &cli.Command{
type apiIpldStore struct {
ctx context.Context
api api.FullNode
api lapi.FullNode
}
func (ht *apiIpldStore) Context() context.Context {
@ -676,7 +816,7 @@ func (ht *apiIpldStore) Put(ctx context.Context, v interface{}) (cid.Cid, error)
panic("No mutations allowed")
}
func handleAmt(ctx context.Context, api api.FullNode, r cid.Cid) error {
func handleAmt(ctx context.Context, api lapi.FullNode, r cid.Cid) error {
s := &apiIpldStore{ctx, api}
mp, err := adt.AsArray(s, r)
if err != nil {
@ -689,7 +829,7 @@ func handleAmt(ctx context.Context, api api.FullNode, r cid.Cid) error {
})
}
func handleHamtEpoch(ctx context.Context, api api.FullNode, r cid.Cid) error {
func handleHamtEpoch(ctx context.Context, api lapi.FullNode, r cid.Cid) error {
s := &apiIpldStore{ctx, api}
mp, err := adt.AsMap(s, r)
if err != nil {
@ -707,7 +847,7 @@ func handleHamtEpoch(ctx context.Context, api api.FullNode, r cid.Cid) error {
})
}
func handleHamtAddress(ctx context.Context, api api.FullNode, r cid.Cid) error {
func handleHamtAddress(ctx context.Context, api lapi.FullNode, r cid.Cid) error {
s := &apiIpldStore{ctx, api}
mp, err := adt.AsMap(s, r)
if err != nil {
@ -930,13 +1070,20 @@ var chainExportCmd = &cli.Command{
return err
}
var last bool
for b := range stream {
last = len(b) == 0
_, err := fi.Write(b)
if err != nil {
return err
}
}
if !last {
return xerrors.Errorf("incomplete export (remote connection lost?)")
}
return nil
},
}

15
cli/cmd.go
View File

@ -216,6 +216,13 @@ func GetAPI(ctx *cli.Context) (api.Common, jsonrpc.ClientCloser, error) {
log.Errorf("repoType type does not match the type of repo.RepoType")
}
if tn, ok := ctx.App.Metadata["testnode-storage"]; ok {
return tn.(api.StorageMiner), func() {}, nil
}
if tn, ok := ctx.App.Metadata["testnode-full"]; ok {
return tn.(api.FullNode), func() {}, nil
}
addr, headers, err := GetRawAPI(ctx, t)
if err != nil {
return nil, nil, err
@ -225,6 +232,10 @@ func GetAPI(ctx *cli.Context) (api.Common, jsonrpc.ClientCloser, error) {
}
func GetFullNodeAPI(ctx *cli.Context) (api.FullNode, jsonrpc.ClientCloser, error) {
if tn, ok := ctx.App.Metadata["testnode-full"]; ok {
return tn.(api.FullNode), func() {}, nil
}
addr, headers, err := GetRawAPI(ctx, repo.FullNode)
if err != nil {
return nil, nil, err
@ -234,6 +245,10 @@ func GetFullNodeAPI(ctx *cli.Context) (api.FullNode, jsonrpc.ClientCloser, error
}
func GetStorageMinerAPI(ctx *cli.Context, opts ...jsonrpc.Option) (api.StorageMiner, jsonrpc.ClientCloser, error) {
if tn, ok := ctx.App.Metadata["testnode-storage"]; ok {
return tn.(api.StorageMiner), func() {}, nil
}
addr, headers, err := GetRawAPI(ctx, repo.StorageMiner)
if err != nil {
return nil, nil, err

69
cli/mpool.go
View File

@ -164,14 +164,6 @@ var mpoolSub = &cli.Command{
},
}
type statBucket struct {
msgs map[uint64]*types.SignedMessage
}
type mpStat struct {
addr string
past, cur, future uint64
}
var mpoolStat = &cli.Command{
Name: "stat",
Usage: "print mempool stats",
@ -180,6 +172,11 @@ var mpoolStat = &cli.Command{
Name: "local",
Usage: "print stats for addresses in local wallet only",
},
&cli.IntFlag{
Name: "basefee-lookback",
Usage: "number of blocks to look back for minimum basefee",
Value: 60,
},
},
Action: func(cctx *cli.Context) error {
api, closer, err := GetFullNodeAPI(cctx)
@ -194,6 +191,20 @@ var mpoolStat = &cli.Command{
if err != nil {
return xerrors.Errorf("getting chain head: %w", err)
}
currBF := ts.Blocks()[0].ParentBaseFee
minBF := currBF
{
currTs := ts
for i := 0; i < cctx.Int("basefee-lookback"); i++ {
currTs, err = api.ChainGetTipSet(ctx, currTs.Parents())
if err != nil {
return xerrors.Errorf("walking chain: %w", err)
}
if newBF := currTs.Blocks()[0].ParentBaseFee; newBF.LessThan(minBF) {
minBF = newBF
}
}
}
var filter map[address.Address]struct{}
if cctx.Bool("local") {
@ -214,8 +225,16 @@ var mpoolStat = &cli.Command{
return err
}
buckets := map[address.Address]*statBucket{}
type statBucket struct {
msgs map[uint64]*types.SignedMessage
}
type mpStat struct {
addr string
past, cur, future uint64
belowCurr, belowPast uint64
}
buckets := map[address.Address]*statBucket{}
for _, v := range msgs {
if filter != nil {
if _, has := filter[v.Message.From]; !has {
@ -252,23 +271,27 @@ var mpoolStat = &cli.Command{
cur++
}
past := uint64(0)
future := uint64(0)
var s mpStat
s.addr = a.String()
for _, m := range bkt.msgs {
if m.Message.Nonce < act.Nonce {
past++
s.past++
} else if m.Message.Nonce > cur {
s.future++
} else {
s.cur++
}
if m.Message.Nonce > cur {
future++
if m.Message.GasFeeCap.LessThan(currBF) {
s.belowCurr++
}
if m.Message.GasFeeCap.LessThan(minBF) {
s.belowPast++
}
}
out = append(out, mpStat{
addr: a.String(),
past: past,
cur: cur - act.Nonce,
future: future,
})
out = append(out, s)
}
sort.Slice(out, func(i, j int) bool {
@ -281,12 +304,14 @@ var mpoolStat = &cli.Command{
total.past += stat.past
total.cur += stat.cur
total.future += stat.future
total.belowCurr += stat.belowCurr
total.belowPast += stat.belowPast
fmt.Printf("%s: past: %d, cur: %d, future: %d\n", stat.addr, stat.past, stat.cur, stat.future)
fmt.Printf("%s: Nonce past: %d, cur: %d, future: %d; FeeCap cur: %d, min-%d: %d \n", stat.addr, stat.past, stat.cur, stat.future, stat.belowCurr, cctx.Int("basefee-lookback"), stat.belowPast)
}
fmt.Println("-----")
fmt.Printf("total: past: %d, cur: %d, future: %d\n", total.past, total.cur, total.future)
fmt.Printf("total: Nonce past: %d, cur: %d, future: %d; FeeCap cur: %d, min-%d: %d \n", total.past, total.cur, total.future, total.belowCurr, cctx.Int("basefee-lookback"), total.belowPast)
return nil
},

290
cmd/lotus-shed/datastore.go Normal file
View File

@ -0,0 +1,290 @@
package main
import (
"encoding/json"
"fmt"
"os"
"strings"
"github.com/docker/go-units"
"github.com/ipfs/go-datastore"
dsq "github.com/ipfs/go-datastore/query"
logging "github.com/ipfs/go-log"
"github.com/polydawn/refmt/cbor"
"github.com/urfave/cli/v2"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/repo"
)
var datastoreCmd = &cli.Command{
Name: "datastore",
Description: "access node datastores directly",
Subcommands: []*cli.Command{
datastoreBackupCmd,
datastoreListCmd,
datastoreGetCmd,
},
}
var datastoreListCmd = &cli.Command{
Name: "list",
Description: "list datastore keys",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "repo-type",
Usage: "node type (1 - full, 2 - storage, 3 - worker)",
Value: 1,
},
&cli.BoolFlag{
Name: "top-level",
Usage: "only print top-level keys",
},
&cli.StringFlag{
Name: "get-enc",
Usage: "print values [esc/hex/cbor]",
},
},
ArgsUsage: "[namespace prefix]",
Action: func(cctx *cli.Context) error {
logging.SetLogLevel("badger", "ERROR") // nolint:errcheck
r, err := repo.NewFS(cctx.String("repo"))
if err != nil {
return xerrors.Errorf("opening fs repo: %w", err)
}
exists, err := r.Exists()
if err != nil {
return err
}
if !exists {
return xerrors.Errorf("lotus repo doesn't exist")
}
lr, err := r.Lock(repo.RepoType(cctx.Int("repo-type")))
if err != nil {
return err
}
defer lr.Close() //nolint:errcheck
ds, err := lr.Datastore(datastore.NewKey(cctx.Args().First()).String())
if err != nil {
return err
}
genc := cctx.String("get-enc")
q, err := ds.Query(dsq.Query{
Prefix: datastore.NewKey(cctx.Args().Get(1)).String(),
KeysOnly: genc == "",
})
if err != nil {
return xerrors.Errorf("datastore query: %w", err)
}
defer q.Close() //nolint:errcheck
printKv := kvPrinter(cctx.Bool("top-level"), genc)
for res := range q.Next() {
if err := printKv(res.Key, res.Value); err != nil {
return err
}
}
return nil
},
}
var datastoreGetCmd = &cli.Command{
Name: "get",
Description: "list datastore keys",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "repo-type",
Usage: "node type (1 - full, 2 - storage, 3 - worker)",
Value: 1,
},
&cli.StringFlag{
Name: "enc",
Usage: "encoding (esc/hex/cbor)",
Value: "esc",
},
},
ArgsUsage: "[namespace key]",
Action: func(cctx *cli.Context) error {
logging.SetLogLevel("badger", "ERROR") // nolint:errchec
r, err := repo.NewFS(cctx.String("repo"))
if err != nil {
return xerrors.Errorf("opening fs repo: %w", err)
}
exists, err := r.Exists()
if err != nil {
return err
}
if !exists {
return xerrors.Errorf("lotus repo doesn't exist")
}
lr, err := r.Lock(repo.RepoType(cctx.Int("repo-type")))
if err != nil {
return err
}
defer lr.Close() //nolint:errcheck
ds, err := lr.Datastore(datastore.NewKey(cctx.Args().First()).String())
if err != nil {
return err
}
val, err := ds.Get(datastore.NewKey(cctx.Args().Get(1)))
if err != nil {
return xerrors.Errorf("get: %w", err)
}
return printVal(cctx.String("enc"), val)
},
}
var datastoreBackupCmd = &cli.Command{
Name: "backup",
Description: "manage datastore backups",
Subcommands: []*cli.Command{
datastoreBackupStatCmd,
datastoreBackupListCmd,
},
}
var datastoreBackupStatCmd = &cli.Command{
Name: "stat",
Description: "validate and print info about datastore backup",
ArgsUsage: "[file]",
Action: func(cctx *cli.Context) error {
if cctx.Args().Len() != 1 {
return xerrors.Errorf("expected 1 argument")
}
f, err := os.Open(cctx.Args().First())
if err != nil {
return xerrors.Errorf("opening backup file: %w", err)
}
defer f.Close() // nolint:errcheck
var keys, kbytes, vbytes uint64
err = backupds.ReadBackup(f, func(key datastore.Key, value []byte) error {
keys++
kbytes += uint64(len(key.String()))
vbytes += uint64(len(value))
return nil
})
if err != nil {
return err
}
fmt.Println("Keys: ", keys)
fmt.Println("Key bytes: ", units.BytesSize(float64(kbytes)))
fmt.Println("Value bytes: ", units.BytesSize(float64(vbytes)))
return err
},
}
var datastoreBackupListCmd = &cli.Command{
Name: "list",
Description: "list data in a backup",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "top-level",
Usage: "only print top-level keys",
},
&cli.StringFlag{
Name: "get-enc",
Usage: "print values [esc/hex/cbor]",
},
},
ArgsUsage: "[file]",
Action: func(cctx *cli.Context) error {
if cctx.Args().Len() != 1 {
return xerrors.Errorf("expected 1 argument")
}
f, err := os.Open(cctx.Args().First())
if err != nil {
return xerrors.Errorf("opening backup file: %w", err)
}
defer f.Close() // nolint:errcheck
printKv := kvPrinter(cctx.Bool("top-level"), cctx.String("get-enc"))
err = backupds.ReadBackup(f, func(key datastore.Key, value []byte) error {
return printKv(key.String(), value)
})
if err != nil {
return err
}
return err
},
}
func kvPrinter(toplevel bool, genc string) func(sk string, value []byte) error {
seen := map[string]struct{}{}
return func(s string, value []byte) error {
if toplevel {
k := datastore.NewKey(datastore.NewKey(s).List()[0])
if k.Type() != "" {
s = k.Type()
} else {
s = k.String()
}
_, has := seen[s]
if has {
return nil
}
seen[s] = struct{}{}
}
s = fmt.Sprintf("%q", s)
s = strings.Trim(s, "\"")
fmt.Println(s)
if genc != "" {
fmt.Print("\t")
if err := printVal(genc, value); err != nil {
return err
}
}
return nil
}
}
func printVal(enc string, val []byte) error {
switch enc {
case "esc":
s := fmt.Sprintf("%q", string(val))
s = strings.Trim(s, "\"")
fmt.Println(s)
case "hex":
fmt.Printf("%x\n", val)
case "cbor":
var out interface{}
if err := cbor.Unmarshal(cbor.DecodeOptions{}, val, &out); err != nil {
return xerrors.Errorf("unmarshaling cbor: %w", err)
}
s, err := json.Marshal(&out)
if err != nil {
return xerrors.Errorf("remarshaling as json: %w", err)
}
fmt.Println(string(s))
default:
return xerrors.New("unknown encoding")
}
return nil
}

1
cmd/lotus-shed/main.go
View File

@ -40,6 +40,7 @@ func main() {
serveDealStatsCmd,
syncCmd,
stateTreePruneCmd,
datastoreCmd,
}
app := &cli.App{

76
cmd/lotus-storage-miner/allinfo_test.go Normal file
View File

@ -0,0 +1,76 @@
package main
import (
"flag"
"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/lib/lotuslog"
"github.com/filecoin-project/lotus/node/repo"
builder "github.com/filecoin-project/lotus/node/test"
)
func TestMinerAllInfo(t *testing.T) {
if testing.Short() {
t.Skip("skipping test in short mode")
}
_ = logging.SetLogLevel("*", "INFO")
policy.SetConsensusMinerMinPower(abi.NewStoragePower(2048))
policy.SetSupportedProofTypes(abi.RegisteredSealProof_StackedDrg2KiBV1)
policy.SetMinVerifiedDealSize(abi.NewStoragePower(256))
_test = true
lotuslog.SetupLogLevels()
logging.SetLogLevel("miner", "ERROR")
logging.SetLogLevel("chainstore", "ERROR")
logging.SetLogLevel("chain", "ERROR")
logging.SetLogLevel("sub", "ERROR")
logging.SetLogLevel("storageminer", "ERROR")
oldDelay := policy.GetPreCommitChallengeDelay()
policy.SetPreCommitChallengeDelay(5)
t.Cleanup(func() {
policy.SetPreCommitChallengeDelay(oldDelay)
})
var n []test.TestNode
var sn []test.TestStorageNode
run := func(t *testing.T) {
app := cli.NewApp()
app.Metadata = map[string]interface{}{
"repoType": repo.StorageMiner,
"testnode-full": n[0],
"testnode-storage": sn[0],
}
build.RunningNodeType = build.NodeMiner
cctx := cli.NewContext(app, flag.NewFlagSet("", flag.ContinueOnError), nil)
require.NoError(t, infoAllCmd.Action(cctx))
}
bp := func(t *testing.T, nFull int, storage []test.StorageMiner) ([]test.TestNode, []test.TestStorageNode) {
n, sn = builder.Builder(t, nFull, storage)
t.Run("pre-info-all", run)
return n, sn
}
test.TestDealFlow(t, bp, time.Second, false, false)
t.Run("post-info-all", run)
}

14
cmd/lotus-storage-miner/backup.go Normal file
View File

@ -0,0 +1,14 @@
package main
import (
"github.com/urfave/cli/v2"
"github.com/filecoin-project/go-jsonrpc"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/filecoin-project/lotus/node/repo"
)
var backupCmd = lcli.BackupCmd(FlagMinerRepo, repo.StorageMiner, func(cctx *cli.Context) (lcli.BackupAPI, jsonrpc.ClientCloser, error) {
return lcli.GetStorageMinerAPI(cctx)
})

10
cmd/lotus-storage-miner/info_all.go
View File

@ -10,6 +10,8 @@ import (
lcli "github.com/filecoin-project/lotus/cli"
)
var _test = false
var infoAllCmd = &cli.Command{
Name: "all",
Usage: "dump all related miner info",
@ -150,9 +152,11 @@ var infoAllCmd = &cli.Command{
}
}
fmt.Println("\n#: Goroutines")
if err := lcli.PprofGoroutines.Action(cctx); err != nil {
return err
if !_test {
fmt.Println("\n#: Goroutines")
if err := lcli.PprofGoroutines.Action(cctx); err != nil {
return err
}
}
return nil

3
cmd/lotus-storage-miner/init.go
View File

@ -115,6 +115,9 @@ var initCmd = &cli.Command{
Usage: "select which address to send actor creation message from",
},
},
Subcommands: []*cli.Command{
initRestoreCmd,
},
Action: func(cctx *cli.Context) error {
log.Info("Initializing lotus miner")

274
cmd/lotus-storage-miner/init_restore.go Normal file
View File

@ -0,0 +1,274 @@
package main
import (
"encoding/json"
"io/ioutil"
"os"
"github.com/docker/go-units"
"github.com/ipfs/go-datastore"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/mitchellh/go-homedir"
"github.com/urfave/cli/v2"
"golang.org/x/xerrors"
"gopkg.in/cheggaaa/pb.v1"
"github.com/filecoin-project/go-address"
paramfetch "github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/types"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/repo"
)
var initRestoreCmd = &cli.Command{
Name: "restore",
Usage: "Initialize a lotus miner repo from a backup",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "nosync",
Usage: "don't check full-node sync status",
},
&cli.StringFlag{
Name: "config",
Usage: "config file (config.toml)",
},
&cli.StringFlag{
Name: "storage-config",
Usage: "storage paths config (storage.json)",
},
},
ArgsUsage: "[backupFile]",
Action: func(cctx *cli.Context) error {
log.Info("Initializing lotus miner using a backup")
if cctx.Args().Len() != 1 {
return xerrors.Errorf("expected 1 argument")
}
log.Info("Trying to connect to full node RPC")
api, closer, err := lcli.GetFullNodeAPI(cctx) // TODO: consider storing full node address in config
if err != nil {
return err
}
defer closer()
log.Info("Checking full node version")
ctx := lcli.ReqContext(cctx)
v, err := api.Version(ctx)
if err != nil {
return err
}
if !v.APIVersion.EqMajorMinor(build.FullAPIVersion) {
return xerrors.Errorf("Remote API version didn't match (expected %s, remote %s)", build.FullAPIVersion, v.APIVersion)
}
if !cctx.Bool("nosync") {
if err := lcli.SyncWait(ctx, api); err != nil {
return xerrors.Errorf("sync wait: %w", err)
}
}
bf, err := homedir.Expand(cctx.Args().First())
if err != nil {
return xerrors.Errorf("expand backup file path: %w", err)
}
st, err := os.Stat(bf)
if err != nil {
return xerrors.Errorf("stat backup file (%s): %w", bf, err)
}
f, err := os.Open(bf)
if err != nil {
return xerrors.Errorf("opening backup file: %w", err)
}
defer f.Close() // nolint:errcheck
log.Info("Checking if repo exists")
repoPath := cctx.String(FlagMinerRepo)
r, err := repo.NewFS(repoPath)
if err != nil {
return err
}
ok, err := r.Exists()
if err != nil {
return err
}
if ok {
return xerrors.Errorf("repo at '%s' is already initialized", cctx.String(FlagMinerRepo))
}
log.Info("Initializing repo")
if err := r.Init(repo.StorageMiner); err != nil {
return err
}
lr, err := r.Lock(repo.StorageMiner)
if err != nil {
return err
}
defer lr.Close() //nolint:errcheck
if cctx.IsSet("config") {
log.Info("Restoring config")
cf, err := homedir.Expand(cctx.String("config"))
if err != nil {
return xerrors.Errorf("expanding config path: %w", err)
}
_, err = os.Stat(cf)
if err != nil {
return xerrors.Errorf("stat config file (%s): %w", cf, err)
}
var cerr error
err = lr.SetConfig(func(raw interface{}) {
rcfg, ok := raw.(*config.StorageMiner)
if !ok {
cerr = xerrors.New("expected miner config")
return
}
ff, err := config.FromFile(cf, rcfg)
if err != nil {
cerr = xerrors.Errorf("loading config: %w", err)
return
}
*rcfg = *ff.(*config.StorageMiner)
})
if cerr != nil {
return cerr
}
if err != nil {
return xerrors.Errorf("setting config: %w", err)
}
} else {
log.Warn("--config NOT SET, WILL USE DEFAULT VALUES")
}
if cctx.IsSet("storage-config") {
log.Info("Restoring storage path config")
cf, err := homedir.Expand(cctx.String("storage-config"))
if err != nil {
return xerrors.Errorf("expanding storage config path: %w", err)
}
cfb, err := ioutil.ReadFile(cf)
if err != nil {
return xerrors.Errorf("reading storage config: %w", err)
}
var cerr error
err = lr.SetStorage(func(scfg *stores.StorageConfig) {
cerr = json.Unmarshal(cfb, scfg)
})
if cerr != nil {
return xerrors.Errorf("unmarshalling storage config: %w", cerr)
}
if err != nil {
return xerrors.Errorf("setting storage config: %w", err)
}
} else {
log.Warn("--storage-config NOT SET. NO SECTOR PATHS WILL BE CONFIGURED")
}
log.Info("Restoring metadata backup")
mds, err := lr.Datastore("/metadata")
if err != nil {
return err
}
bar := pb.New64(st.Size())
br := bar.NewProxyReader(f)
bar.ShowTimeLeft = true
bar.ShowPercent = true
bar.ShowSpeed = true
bar.Units = pb.U_BYTES
bar.Start()
err = backupds.RestoreInto(br, mds)
bar.Finish()
if err != nil {
return xerrors.Errorf("restoring metadata: %w", err)
}
log.Info("Checking actor metadata")
abytes, err := mds.Get(datastore.NewKey("miner-address"))
if err != nil {
return xerrors.Errorf("getting actor address from metadata datastore: %w", err)
}
maddr, err := address.NewFromBytes(abytes)
if err != nil {
return xerrors.Errorf("parsing actor address: %w", err)
}
log.Info("ACTOR ADDRESS: ", maddr.String())
mi, err := api.StateMinerInfo(ctx, maddr, types.EmptyTSK)
if err != nil {
return xerrors.Errorf("getting miner info: %w", err)
}
log.Info("SECTOR SIZE: ", units.BytesSize(float64(mi.SectorSize)))
wk, err := api.StateAccountKey(ctx, mi.Worker, types.EmptyTSK)
if err != nil {
return xerrors.Errorf("resolving worker key: %w", err)
}
has, err := api.WalletHas(ctx, wk)
if err != nil {
return xerrors.Errorf("checking worker address: %w", err)
}
if !has {
return xerrors.Errorf("worker address %s for miner actor %s not present in full node wallet", mi.Worker, maddr)
}
log.Info("Checking proof parameters")
if err := paramfetch.GetParams(ctx, build.ParametersJSON(), uint64(mi.SectorSize)); err != nil {
return xerrors.Errorf("fetching proof parameters: %w", err)
}
log.Info("Initializing libp2p identity")
p2pSk, err := makeHostKey(lr)
if err != nil {
return xerrors.Errorf("make host key: %w", err)
}
peerid, err := peer.IDFromPrivateKey(p2pSk)
if err != nil {
return xerrors.Errorf("peer ID from private key: %w", err)
}
log.Info("Configuring miner actor")
if err := configureStorageMiner(ctx, api, maddr, peerid, big.Zero()); err != nil {
return err
}
return nil
},
}

1
cmd/lotus-storage-miner/main.go
View File

@ -35,6 +35,7 @@ func main() {
runCmd,
stopCmd,
configCmd,
backupCmd,
lcli.WithCategory("chain", actorCmd),
lcli.WithCategory("chain", infoCmd),
lcli.WithCategory("market", storageDealsCmd),

14
cmd/lotus/backup.go Normal file
View File

@ -0,0 +1,14 @@
package main
import (
"github.com/urfave/cli/v2"
"github.com/filecoin-project/go-jsonrpc"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/filecoin-project/lotus/node/repo"
)
var backupCmd = lcli.BackupCmd("repo", repo.FullNode, func(cctx *cli.Context) (lcli.BackupAPI, jsonrpc.ClientCloser, error) {
return lcli.GetFullNodeAPI(cctx)
})

1
cmd/lotus/main.go
View File

@ -22,6 +22,7 @@ func main() {
local := []*cli.Command{
DaemonCmd,
backupCmd,
}
if AdvanceBlockCmd != nil {
local = append(local, AdvanceBlockCmd)

383
documentation/en/api-methods.md
View File

File diff suppressed because it is too large Load Diff

1
go.mod
View File

@ -112,6 +112,7 @@ require (
github.com/multiformats/go-multibase v0.0.3
github.com/multiformats/go-multihash v0.0.14
github.com/opentracing/opentracing-go v1.2.0
github.com/polydawn/refmt v0.0.0-20190809202753-05966cbd336a
github.com/raulk/clock v1.1.0
github.com/stretchr/testify v1.6.1
github.com/supranational/blst v0.1.1

189
lib/backupds/datastore.go Normal file
View File

@ -0,0 +1,189 @@
package backupds
import (
"crypto/sha256"
"io"
"sync"
logging "github.com/ipfs/go-log/v2"
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/query"
)
var log = logging.Logger("backupds")
type Datastore struct {
child datastore.Batching
backupLk sync.RWMutex
}
func Wrap(child datastore.Batching) *Datastore {
return &Datastore{
child: child,
}
}
// Writes a datastore dump into the provided writer as
// [array(*) of [key, value] tuples, checksum]
func (d *Datastore) Backup(out io.Writer) error {
scratch := make([]byte, 9)
if err := cbg.WriteMajorTypeHeaderBuf(scratch, out, cbg.MajArray, 2); err != nil {
return xerrors.Errorf("writing tuple header: %w", err)
}
hasher := sha256.New()
hout := io.MultiWriter(hasher, out)
// write KVs
{
// write indefinite length array header
if _, err := hout.Write([]byte{0x9f}); err != nil {
return xerrors.Errorf("writing header: %w", err)
}
d.backupLk.Lock()
defer d.backupLk.Unlock()
log.Info("Starting datastore backup")
defer log.Info("Datastore backup done")
qr, err := d.child.Query(query.Query{})
if err != nil {
return xerrors.Errorf("query: %w", err)
}
defer func() {
if err := qr.Close(); err != nil {
log.Errorf("query close error: %+v", err)
return
}
}()
for result := range qr.Next() {
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajArray, 2); err != nil {
return xerrors.Errorf("writing tuple header: %w", err)
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajByteString, uint64(len([]byte(result.Key)))); err != nil {
return xerrors.Errorf("writing key header: %w", err)
}
if _, err := hout.Write([]byte(result.Key)[:]); err != nil {
return xerrors.Errorf("writing key: %w", err)
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajByteString, uint64(len(result.Value))); err != nil {
return xerrors.Errorf("writing value header: %w", err)
}
if _, err := hout.Write(result.Value[:]); err != nil {
return xerrors.Errorf("writing value: %w", err)
}
}
// array break
if _, err := hout.Write([]byte{0xff}); err != nil {
return xerrors.Errorf("writing array 'break': %w", err)
}
}
// Write the checksum
{
sum := hasher.Sum(nil)
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajByteString, uint64(len(sum))); err != nil {
return xerrors.Errorf("writing checksum header: %w", err)
}
if _, err := hout.Write(sum[:]); err != nil {
return xerrors.Errorf("writing checksum: %w", err)
}
}
return nil
}
// proxy
func (d *Datastore) Get(key datastore.Key) (value []byte, err error) {
return d.child.Get(key)
}
func (d *Datastore) Has(key datastore.Key) (exists bool, err error) {
return d.child.Has(key)
}
func (d *Datastore) GetSize(key datastore.Key) (size int, err error) {
return d.child.GetSize(key)
}
func (d *Datastore) Query(q query.Query) (query.Results, error) {
return d.child.Query(q)
}
func (d *Datastore) Put(key datastore.Key, value []byte) error {
d.backupLk.RLock()
defer d.backupLk.RUnlock()
return d.child.Put(key, value)
}
func (d *Datastore) Delete(key datastore.Key) error {
d.backupLk.RLock()
defer d.backupLk.RUnlock()
return d.child.Delete(key)
}
func (d *Datastore) Sync(prefix datastore.Key) error {
d.backupLk.RLock()
defer d.backupLk.RUnlock()
return d.child.Sync(prefix)
}
func (d *Datastore) Close() error {
d.backupLk.RLock()
defer d.backupLk.RUnlock()
return d.child.Close()
}
func (d *Datastore) Batch() (datastore.Batch, error) {
b, err := d.child.Batch()
if err != nil {
return nil, err
}
return &bbatch{
b: b,
rlk: d.backupLk.RLocker(),
}, nil
}
type bbatch struct {
b datastore.Batch
rlk sync.Locker
}
func (b *bbatch) Put(key datastore.Key, value []byte) error {
return b.b.Put(key, value)
}
func (b *bbatch) Delete(key datastore.Key) error {
return b.b.Delete(key)
}
func (b *bbatch) Commit() error {
b.rlk.Lock()
defer b.rlk.Unlock()
return b.b.Commit()
}
var _ datastore.Batch = &bbatch{}
var _ datastore.Batching = &Datastore{}

100
lib/backupds/read.go Normal file
View File

@ -0,0 +1,100 @@
package backupds
import (
"bytes"
"crypto/sha256"
"io"
"github.com/ipfs/go-datastore"
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
)
func ReadBackup(r io.Reader, cb func(key datastore.Key, value []byte) error) error {
scratch := make([]byte, 9)
if _, err := r.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading array header: %w", err)
}
if scratch[0] != 0x82 {
return xerrors.Errorf("expected array(2) header byte 0x82, got %x", scratch[0])
}
hasher := sha256.New()
hr := io.TeeReader(r, hasher)
if _, err := hr.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading array header: %w", err)
}
if scratch[0] != 0x9f {
return xerrors.Errorf("expected indefinite length array header byte 0x9f, got %x", scratch[0])
}
for {
if _, err := hr.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading tuple header: %w", err)
}
if scratch[0] == 0xff {
break
}
if scratch[0] != 0x82 {
return xerrors.Errorf("expected array(2) header 0x82, got %x", scratch[0])
}
keyb, err := cbg.ReadByteArray(hr, 1<<40)
if err != nil {
return xerrors.Errorf("reading key: %w", err)
}
key := datastore.NewKey(string(keyb))
value, err := cbg.ReadByteArray(hr, 1<<40)
if err != nil {
return xerrors.Errorf("reading value: %w", err)
}
if err := cb(key, value); err != nil {
return err
}
}
sum := hasher.Sum(nil)
expSum, err := cbg.ReadByteArray(r, 32)
if err != nil {
return xerrors.Errorf("reading expected checksum: %w", err)
}
if !bytes.Equal(sum, expSum) {
return xerrors.Errorf("checksum didn't match; expected %x, got %x", expSum, sum)
}
return nil
}
func RestoreInto(r io.Reader, dest datastore.Batching) error {
batch, err := dest.Batch()
if err != nil {
return xerrors.Errorf("creating batch: %w", err)
}
err = ReadBackup(r, func(key datastore.Key, value []byte) error {
if err := batch.Put(key, value); err != nil {
return xerrors.Errorf("put key: %w", err)
}
return nil
})
if err != nil {
return xerrors.Errorf("reading backup: %w", err)
}
if err := batch.Commit(); err != nil {
return xerrors.Errorf("committing batch: %w", err)
}
return nil
}

67
node/impl/backup.go Normal file
View File

@ -0,0 +1,67 @@
package impl
import (
"os"
"path/filepath"
"strings"
"github.com/mitchellh/go-homedir"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
func backup(mds dtypes.MetadataDS, fpath string) error {
bb, ok := os.LookupEnv("LOTUS_BACKUP_BASE_PATH")
if !ok {
return xerrors.Errorf("LOTUS_BACKUP_BASE_PATH env var not set")
}
bds, ok := mds.(*backupds.Datastore)
if !ok {
return xerrors.Errorf("expected a backup datastore")
}
bb, err := homedir.Expand(bb)
if err != nil {
return xerrors.Errorf("expanding base path: %w", err)
}
bb, err = filepath.Abs(bb)
if err != nil {
return xerrors.Errorf("getting absolute base path: %w", err)
}
fpath, err = homedir.Expand(fpath)
if err != nil {
return xerrors.Errorf("expanding file path: %w", err)
}
fpath, err = filepath.Abs(fpath)
if err != nil {
return xerrors.Errorf("getting absolute file path: %w", err)
}
if !strings.HasPrefix(fpath, bb) {
return xerrors.Errorf("backup file name (%s) must be inside base path (%s)", fpath, bb)
}
out, err := os.OpenFile(fpath, os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return xerrors.Errorf("open %s: %w", fpath, err)
}
if err := bds.Backup(out); err != nil {
if cerr := out.Close(); cerr != nil {
log.Errorw("error closing backup file while handling backup error", "closeErr", cerr, "backupErr", err)
}
return xerrors.Errorf("backup error: %w", err)
}
if err := out.Close(); err != nil {
return xerrors.Errorf("closing backup file: %w", err)
}
return nil
}

6
node/impl/common/common.go
View File

@ -121,6 +121,12 @@ func (a *CommonAPI) NetFindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo,
func (a *CommonAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error) {
autonat := a.RawHost.(*basichost.BasicHost).AutoNat
if autonat == nil {
return api.NatInfo{
Reachability: network.ReachabilityUnknown,
}, nil
}
var maddr string
if autonat.Status() == network.ReachabilityPublic {
pa, err := autonat.PublicAddr()

9
node/impl/full.go
View File

@ -1,6 +1,8 @@
package impl
import (
"context"
logging "github.com/ipfs/go-log/v2"
"github.com/filecoin-project/lotus/api"
@ -9,6 +11,7 @@ import (
"github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/node/impl/market"
"github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
var log = logging.Logger("node")
@ -26,6 +29,12 @@ type FullNodeAPI struct {
full.WalletAPI
full.SyncAPI
full.BeaconAPI
DS dtypes.MetadataDS
}
func (n *FullNodeAPI) CreateBackup(ctx context.Context, fpath string) error {
return backup(n.DS, fpath)
}
var _ api.FullNode = &FullNodeAPI{}

30
node/impl/full/chain.go
View File

@ -509,15 +509,11 @@ func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, skipo
r, w := io.Pipe()
out := make(chan []byte)
go func() {
defer w.Close() //nolint:errcheck // it is a pipe
bw := bufio.NewWriterSize(w, 1<<20)
defer bw.Flush() //nolint:errcheck // it is a write to a pipe
if err := a.Chain.Export(ctx, ts, nroots, skipoldmsgs, bw); err != nil {
log.Errorf("chain export call failed: %s", err)
return
}
err := a.Chain.Export(ctx, ts, nroots, skipoldmsgs, bw)
bw.Flush() //nolint:errcheck // it is a write to a pipe
w.CloseWithError(err) //nolint:errcheck // it is a pipe
}()
go func() {
@ -529,13 +525,23 @@ func (a *ChainAPI) ChainExport(ctx context.Context, nroots abi.ChainEpoch, skipo
log.Errorf("chain export pipe read failed: %s", err)
return
}
select {
case out <- buf[:n]:
case <-ctx.Done():
log.Warnf("export writer failed: %s", ctx.Err())
return
if n > 0 {
select {
case out <- buf[:n]:
case <-ctx.Done():
log.Warnf("export writer failed: %s", ctx.Err())
return
}
}
if err == io.EOF {
// send empty slice to indicate correct eof
select {
case out <- []byte{}:
case <-ctx.Done():
log.Warnf("export writer failed: %s", ctx.Err())
return
}
return
}
}

10
node/impl/storminer.go
View File

@ -8,18 +8,18 @@ import (
"strconv"
"time"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-state-types/big"
"github.com/ipfs/go-cid"
"github.com/libp2p/go-libp2p-core/host"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-fil-markets/piecestore"
retrievalmarket "github.com/filecoin-project/go-fil-markets/retrievalmarket"
storagemarket "github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/go-jsonrpc/auth"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
@ -56,6 +56,8 @@ type StorageMinerAPI struct {
DataTransfer dtypes.ProviderDataTransfer
Host host.Host
DS dtypes.MetadataDS
ConsiderOnlineStorageDealsConfigFunc dtypes.ConsiderOnlineStorageDealsConfigFunc
SetConsiderOnlineStorageDealsConfigFunc dtypes.SetConsiderOnlineStorageDealsConfigFunc
ConsiderOnlineRetrievalDealsConfigFunc dtypes.ConsiderOnlineRetrievalDealsConfigFunc
@ -516,4 +518,8 @@ func (sm *StorageMinerAPI) PiecesGetCIDInfo(ctx context.Context, payloadCid cid.
return &ci, nil
}
func (sm *StorageMinerAPI) CreateBackup(ctx context.Context, fpath string) error {
return backup(sm.DS, fpath)
}
var _ api.StorageMiner = &StorageMinerAPI{}

4
node/modules/lp2p/pubsub.go
View File

@ -223,8 +223,8 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
},
AppSpecificWeight: 1,
// This sets the IP colocation threshold to 1 peer per
IPColocationFactorThreshold: 1,
// This sets the IP colocation threshold to 5 peers before we apply penalties
IPColocationFactorThreshold: 5,
IPColocationFactorWeight: -100,
// TODO we want to whitelist IPv6 /64s that belong to datacenters etc
// IPColocationFactorWhitelist: map[string]struct{}{},

8
node/modules/storage.go
View File

@ -6,6 +6,7 @@ import (
"go.uber.org/fx"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
)
@ -27,5 +28,10 @@ func KeyStore(lr repo.LockedRepo) (types.KeyStore, error) {
}
func Datastore(r repo.LockedRepo) (dtypes.MetadataDS, error) {
return r.Datastore("/metadata")
mds, err := r.Datastore("/metadata")
if err != nil {
return nil, err
}
return backupds.Wrap(mds), nil
}

8
node/modules/storageminer.go
View File

@ -44,6 +44,7 @@ import (
paramfetch "github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-storedcounter"
"github.com/filecoin-project/specs-actors/actors/builtin"
sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
@ -468,6 +469,13 @@ func BasicDealFilter(user dtypes.DealFilter) func(onlineOk dtypes.ConsiderOnline
return false, fmt.Sprintf("cannot seal a sector before %s", deal.Proposal.StartEpoch), nil
}
// Reject if it's more than 7 days in the future
// TODO: read from cfg
maxStartEpoch := ht + abi.ChainEpoch(7*builtin.EpochsInDay)
if deal.Proposal.StartEpoch > maxStartEpoch {
return false, fmt.Sprintf("deal start epoch is too far in the future: %s > %s", deal.Proposal.StartEpoch, maxStartEpoch), nil
}
if user != nil {
return user(ctx, deal)
}

12
node/repo/fsrepo.go
View File

@ -226,11 +226,23 @@ func (fsr *FsRepo) Lock(repoType RepoType) (LockedRepo, error) {
}, nil
}
// Like Lock, except datastores will work in read-only mode
func (fsr *FsRepo) LockRO(repoType RepoType) (LockedRepo, error) {
lr, err := fsr.Lock(repoType)
if err != nil {
return nil, err
}
lr.(*fsLockedRepo).readonly = true
return lr, nil
}
type fsLockedRepo struct {
path string
configPath string
repoType RepoType
closer io.Closer
readonly bool
ds map[string]datastore.Batching
dsErr error

18
node/repo/fsrepo_ds.go
View File

@ -14,7 +14,7 @@ import (
ldbopts "github.com/syndtr/goleveldb/leveldb/opt"
)
type dsCtor func(path string) (datastore.Batching, error)
type dsCtor func(path string, readonly bool) (datastore.Batching, error)
var fsDatastores = map[string]dsCtor{
"chain": chainBadgerDs,
@ -26,9 +26,10 @@ var fsDatastores = map[string]dsCtor{
"client": badgerDs, // client specific
}
func chainBadgerDs(path string) (datastore.Batching, error) {
func chainBadgerDs(path string, readonly bool) (datastore.Batching, error) {
opts := badger.DefaultOptions
opts.GcInterval = 0 // disable GC for chain datastore
opts.ReadOnly = readonly
opts.Options = dgbadger.DefaultOptions("").WithTruncate(true).
WithValueThreshold(1 << 10)
@ -36,23 +37,26 @@ func chainBadgerDs(path string) (datastore.Batching, error) {
return badger.NewDatastore(path, &opts)
}
func badgerDs(path string) (datastore.Batching, error) {
func badgerDs(path string, readonly bool) (datastore.Batching, error) {
opts := badger.DefaultOptions
opts.ReadOnly = readonly
opts.Options = dgbadger.DefaultOptions("").WithTruncate(true).
WithValueThreshold(1 << 10)
return badger.NewDatastore(path, &opts)
}
func levelDs(path string) (datastore.Batching, error) {
func levelDs(path string, readonly bool) (datastore.Batching, error) {
return levelds.NewDatastore(path, &levelds.Options{
Compression: ldbopts.NoCompression,
NoSync: false,
Strict: ldbopts.StrictAll,
ReadOnly: readonly,
})
}
func (fsr *fsLockedRepo) openDatastores() (map[string]datastore.Batching, error) {
func (fsr *fsLockedRepo) openDatastores(readonly bool) (map[string]datastore.Batching, error) {
if err := os.MkdirAll(fsr.join(fsDatastore), 0755); err != nil {
return nil, xerrors.Errorf("mkdir %s: %w", fsr.join(fsDatastore), err)
}
@ -63,7 +67,7 @@ func (fsr *fsLockedRepo) openDatastores() (map[string]datastore.Batching, error)
prefix := datastore.NewKey(p)
// TODO: optimization: don't init datastores we don't need
ds, err := ctor(fsr.join(filepath.Join(fsDatastore, p)))
ds, err := ctor(fsr.join(filepath.Join(fsDatastore, p)), readonly)
if err != nil {
return nil, xerrors.Errorf("opening datastore %s: %w", prefix, err)
}
@ -78,7 +82,7 @@ func (fsr *fsLockedRepo) openDatastores() (map[string]datastore.Batching, error)
func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
fsr.dsOnce.Do(func() {
fsr.ds, fsr.dsErr = fsr.openDatastores()
fsr.ds, fsr.dsErr = fsr.openDatastores(fsr.readonly)
})
if fsr.dsErr != nil {

16
paychmgr/store_test.go
View File

@ -53,8 +53,12 @@ func TestStore(t *testing.T) {
addrs, err = store.ListChannels()
require.NoError(t, err)
require.Len(t, addrs, 2)
require.Contains(t, addrsStrings(addrs), "t0100")
require.Contains(t, addrsStrings(addrs), "t0200")
t0100, err := address.NewIDAddress(100)
require.NoError(t, err)
t0200, err := address.NewIDAddress(200)
require.NoError(t, err)
require.Contains(t, addrs, t0100)
require.Contains(t, addrs, t0200)
// Request vouchers for channel
vouchers, err := store.VouchersForPaych(*ci.Channel)
@ -79,11 +83,3 @@ func TestStore(t *testing.T) {
_, err = store.AllocateLane(tutils.NewIDAddr(t, 300))
require.Equal(t, err, ErrChannelNotTracked)
}
func addrsStrings(addrs []address.Address) []string {
str := make([]string, len(addrs))
for i, a := range addrs {
str[i] = a.String()
}
return str
}