feat: booster-http

Author: dirkmc

Makefile

@@ -89,6 +89,12 @@ boost: $(BUILD_DEPS)
 .PHONY: boost
 BINS+=boost boostx boostd
 
+booster-http: $(BUILD_DEPS)
+	rm -f booster-http
+	$(GOCC) build $(GOFLAGS) -o booster-http ./cmd/booster-http
+.PHONY: booster-http
+BINS+=booster-http
+
 devnet: $(BUILD_DEPS)
 	rm -f devnet
 	$(GOCC) build $(GOFLAGS) -o devnet ./cmd/devnet

api/api.go

@@ -66,6 +66,7 @@ type Boost interface {
 	PiecesListCidInfos(ctx context.Context) ([]cid.Cid, error)                               //perm:read
 	PiecesGetPieceInfo(ctx context.Context, pieceCid cid.Cid) (*piecestore.PieceInfo, error) //perm:read
 	PiecesGetCIDInfo(ctx context.Context, payloadCid cid.Cid) (*piecestore.CIDInfo, error)   //perm:read
+	PiecesGetMaxOffset(ctx context.Context, pieceCid cid.Cid) (uint64, error)                //perm:read
 
 	// MethodGroup: Actor
 	ActorSectorSize(context.Context, address.Address) (abi.SectorSize, error) //perm:read

api/proxy_gen.go

@@ -28,6 +28,10 @@ var runCmd = &cli.Command{
 			Name:  "pprof",
 			Usage: "run pprof web server on localhost:6060",
 		},
+		&cli.BoolFlag{
+			Name:  "nosync",
+			Usage: "dont wait for the full node to sync with the chain",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		if cctx.Bool("pprof") {

cmd/boostd/run.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"os"
+
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/urfave/cli/v2"
+
+	"github.com/filecoin-project/boost/build"
+	cliutil "github.com/filecoin-project/boost/cli/util"
+)
+
+var log = logging.Logger("booster")
+
+const (
+	FlagBoostRepo = "boost-repo"
+)
+
+func main() {
+	app := &cli.App{
+		Name:                 "booster-http",
+		Usage:                "HTTP endpoint for retrieval from Filecoin",
+		EnableBashCompletion: true,
+		Version:              build.UserVersion(),
+		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name:    FlagBoostRepo,
+				EnvVars: []string{"BOOST_PATH"},
+				Usage:   "boost repo path",
+				Value:   "~/.boost",
+			},
+			cliutil.FlagVeryVerbose,
+		},
+		Commands: []*cli.Command{
+			runCmd,
+		},
+	}
+	app.Setup()
+
+	if err := app.Run(os.Args); err != nil {
+		os.Stderr.WriteString("Error: " + err.Error() + "\n")
+	}
+}
+
+func before(cctx *cli.Context) error {
+	_ = logging.SetLogLevel("booster", "INFO")
+
+	if cliutil.IsVeryVerbose {
+		_ = logging.SetLogLevel("booster", "DEBUG")
+	}
+
+	return nil
+}

cmd/booster-http/main.go

@@ -0,0 +1,263 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/filecoin-project/boost/api"
+	bclient "github.com/filecoin-project/boost/api/client"
+	cliutil "github.com/filecoin-project/boost/cli/util"
+	"github.com/filecoin-project/dagstore/mount"
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-fil-markets/piecestore"
+	"github.com/filecoin-project/go-jsonrpc"
+	"github.com/filecoin-project/go-state-types/abi"
+	lapi "github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/api/client"
+	"github.com/filecoin-project/lotus/api/v0api"
+	"github.com/filecoin-project/lotus/api/v1api"
+	lcli "github.com/filecoin-project/lotus/cli"
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
+	"github.com/filecoin-project/lotus/markets/dagstore"
+	"github.com/filecoin-project/lotus/markets/sectoraccessor"
+	lotus_modules "github.com/filecoin-project/lotus/node/modules"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/filecoin-project/lotus/node/repo"
+	"github.com/ipfs/go-cid"
+	"github.com/urfave/cli/v2"
+	"net/http"
+	_ "net/http/pprof"
+	"strings"
+)
+
+var runCmd = &cli.Command{
+	Name:   "run",
+	Usage:  "Start a booster-http process",
+	Before: before,
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "pprof",
+			Usage: "run pprof web server on localhost:6070",
+		},
+		&cli.StringFlag{
+			Name:  "base-path",
+			Usage: "the base path at which to run the web server",
+			Value: "",
+		},
+		&cli.UintFlag{
+			Name:  "port",
+			Usage: "the port the web server listens on",
+			Value: 7777,
+		},
+		&cli.StringFlag{
+			Name:     "api-boost",
+			Usage:    "the endpoint for the boost API",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "api-fullnode",
+			Usage:    "the endpoint for the full node API",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "api-sealer",
+			Usage:    "the endpoint for the sealer API",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "miner",
+			Usage:    "the miner's address",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		if cctx.Bool("pprof") {
+			go func() {
+				err := http.ListenAndServe("localhost:6070", nil)
+				if err != nil {
+					log.Error(err)
+				}
+			}()
+		}
+
+		maddr, err := address.NewFromString(cctx.String("miner"))
+		if err != nil {
+			return fmt.Errorf("parsing miner address %s: %w", cctx.String("miner"), err)
+		}
+
+		// Connect to the Boost API
+		ctx := lcli.ReqContext(cctx)
+		boostApiInfo := cctx.String("api-boost")
+		bapi, bcloser, err := getBoostApi(ctx, boostApiInfo)
+		if err != nil {
+			return fmt.Errorf("getting boost API: %w", err)
+		}
+		defer bcloser()
+
+		// Connect to the full node API
+		fnApiInfo := cctx.String("api-fullnode")
+		fullnodeApi, ncloser, err := getFullNodeApi(ctx, fnApiInfo)
+		defer ncloser()
+
+		// Connect to the sealing API
+		sealingApiInfo := cctx.String("api-sealer")
+		sauth, err := storageAuthWithURL(sealingApiInfo)
+		if err != nil {
+			return fmt.Errorf("parsing sealing API endpoint: %w", err)
+		}
+		sealingService, sealerCloser, err := getMinerApi(ctx, sealingApiInfo)
+		defer sealerCloser()
+
+		// Use an in-memory repo because we don't need any functions
+		// of a real repo, we just need to supply something that satisfies
+		// the LocalStorage interface to the store
+		memRepo := repo.NewMemory(nil)
+		lr, err := memRepo.Lock(repo.StorageMiner)
+		if err != nil {
+			return fmt.Errorf("locking mem repo: %w", err)
+		}
+		defer lr.Close()
+
+		// Create the store interface
+		var urls []string
+		lstor, err := stores.NewLocal(ctx, lr, sealingService, urls)
+		if err != nil {
+			return fmt.Errorf("creating new local store: %w", err)
+		}
+		storage := lotus_modules.RemoteStorage(lstor, sealingService, sauth, sectorstorage.Config{
+			// TODO: Not sure if I need this, or any of the other fields in this struct
+			ParallelFetchLimit: 1,
+		})
+		// Create the piece provider and sector accessors
+		pp := sectorstorage.NewPieceProvider(storage, sealingService, sealingService)
+		sa := sectoraccessor.NewSectorAccessor(dtypes.MinerAddress(maddr), sealingService, pp, fullnodeApi)
+		// Create the server API
+		sapi := serverApi{ctx: ctx, bapi: bapi, sa: sa}
+		server := NewHttpServer(cctx.String("base-path"), cctx.Int("port"), sapi)
+
+		// Start the server
+		log.Infof("Starting booster-http node on port %d with base path '%s'",
+			cctx.Int("port"), cctx.String("base-path"))
+		server.Start(ctx)
+
+		// Monitor for shutdown.
+		<-ctx.Done()
+
+		log.Info("Shutting down...")
+
+		err = server.Stop()
+		if err != nil {
+			return err
+		}
+		log.Info("Graceful shutdown successful")
+
+		// Sync all loggers.
+		_ = log.Sync() //nolint:errcheck
+
+		return nil
+	},
+}
+
+func storageAuthWithURL(apiInfo string) (sectorstorage.StorageAuth, error) {
+	s := strings.Split(apiInfo, ":")
+	if len(s) != 2 {
+		return nil, errors.New("unexpected format of `apiInfo`")
+	}
+	headers := http.Header{}
+	headers.Add("Authorization", "Bearer "+s[0])
+	return sectorstorage.StorageAuth(headers), nil
+}
+
+type serverApi struct {
+	ctx  context.Context
+	bapi api.Boost
+	sa   dagstore.SectorAccessor
+}
+
+var _ HttpServerApi = (*serverApi)(nil)
+
+func (s serverApi) GetMaxPieceOffset(pieceCid cid.Cid) (uint64, error) {
+	return s.bapi.PiecesGetMaxOffset(s.ctx, pieceCid)
+}
+
+func (s serverApi) GetPieceInfo(pieceCID cid.Cid) (*piecestore.PieceInfo, error) {
+	return s.bapi.PiecesGetPieceInfo(s.ctx, pieceCID)
+}
+
+func (s serverApi) IsUnsealed(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (bool, error) {
+	return s.sa.IsUnsealed(ctx, sectorID, offset, length)
+}
+
+func (s serverApi) UnsealSectorAt(ctx context.Context, sectorID abi.SectorNumber, offset abi.UnpaddedPieceSize, length abi.UnpaddedPieceSize) (mount.Reader, error) {
+	return s.sa.UnsealSectorAt(ctx, sectorID, offset, length)
+}
+
+func getBoostApi(ctx context.Context, ai string) (api.Boost, jsonrpc.ClientCloser, error) {
+	ai = strings.TrimPrefix(strings.TrimSpace(ai), "BOOST_API_INFO=")
+	info := cliutil.ParseApiInfo(ai)
+	addr, err := info.DialArgs("v0")
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not get DialArgs: %w", err)
+	}
+
+	log.Infof("Using boost API at %s", addr)
+	api, closer, err := bclient.NewBoostRPCV0(ctx, addr, info.AuthHeader())
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating full node service API: %w", err)
+	}
+
+	return api, closer, nil
+}
+
+func getFullNodeApi(ctx context.Context, ai string) (v1api.FullNode, jsonrpc.ClientCloser, error) {
+	ai = strings.TrimPrefix(strings.TrimSpace(ai), "FULLNODE_API_INFO=")
+	info := cliutil.ParseApiInfo(ai)
+	addr, err := info.DialArgs("v1")
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not get DialArgs: %w", err)
+	}
+
+	log.Infof("Using full node API at %s", addr)
+	api, closer, err := client.NewFullNodeRPCV1(ctx, addr, info.AuthHeader())
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating full node service API: %w", err)
+	}
+
+	v, err := api.Version(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("checking full node service API version: %w", err)
+	}
+
+	if !v.APIVersion.EqMajorMinor(lapi.FullAPIVersion1) {
+		return nil, nil, fmt.Errorf("full node service API version didn't match (expected %s, remote %s)", lapi.FullAPIVersion1, v.APIVersion)
+	}
+
+	return api, closer, nil
+}
+
+func getMinerApi(ctx context.Context, ai string) (v0api.StorageMiner, jsonrpc.ClientCloser, error) {
+	ai = strings.TrimPrefix(strings.TrimSpace(ai), "MINER_API_INFO=")
+	info := cliutil.ParseApiInfo(ai)
+	addr, err := info.DialArgs("v0")
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not get DialArgs: %w", err)
+	}
+
+	log.Infof("Using sealing API at %s", addr)
+	api, closer, err := client.NewStorageMinerRPCV0(ctx, addr, info.AuthHeader())
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating miner service API: %w", err)
+	}
+
+	v, err := api.Version(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("checking miner service API version: %w", err)
+	}
+
+	if !v.APIVersion.EqMajorMinor(lapi.MinerAPIVersion0) {
+		return nil, nil, fmt.Errorf("miner service API version didn't match (expected %s, remote %s)", lapi.MinerAPIVersion0, v.APIVersion)
+	}
+
+	return api, closer, nil
+}