market: Real-world fixes

cmd/curio/debug-ipni.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/ipfs/go-cid"
+	carv2 "github.com/ipld/go-car/v2"
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/curio/market/ipni/chunker"
+)
+
+var testDebugIpniChunks = &cli.Command{
+	Name:  "ipni-piece-chunks",
+	Usage: "generate ipni chunks from a file",
+	Action: func(c *cli.Context) error {
+		ck := chunker.NewInitialChunker()
+
+		f, err := os.Open(c.Args().First())
+		if err != nil {
+			return xerrors.Errorf("opening file: %w", err)
+		}
+		defer f.Close()
+
+		opts := []carv2.Option{carv2.ZeroLengthSectionAsEOF(true)}
+		blockReader, err := carv2.NewBlockReader(bufio.NewReaderSize(f, 4<<20), opts...)
+		if err != nil {
+			return fmt.Errorf("getting block reader over piece: %w", err)
+		}
+
+		blockMetadata, err := blockReader.SkipNext()
+		for err == nil {
+			if err := ck.Accept(blockMetadata.Cid.Hash(), int64(blockMetadata.Offset), blockMetadata.Size+40); err != nil {
+				return xerrors.Errorf("accepting block: %w", err)
+			}
+
+			blockMetadata, err = blockReader.SkipNext()
+		}
+		if !errors.Is(err, io.EOF) {
+			return xerrors.Errorf("reading block: %w", err)
+		}
+
+		_, err = ck.Finish(c.Context, nil, cid.Undef)
+		if err != nil {
+			return xerrors.Errorf("chunking CAR multihash iterator: %w", err)
+		}
+
+		return nil
+	},
+}

cmd/curio/seal.go

@@ -152,7 +152,7 @@ var sealStartCmd = &cli.Command{
 			}
 
 			for _, n := range num {
-				_, err := tx.Exec("insert into sectors_sdr_pipeline (sp_id, sector_number, reg_seal_proof) values ($1, $2, $3)", mid, n, spt)
+				_, err := tx.Exec("insert into sectors_sdr_pipeline (sp_id, sector_number, reg_seal_proof, user_sector_duration_epochs) values ($1, $2, $3, $4)", mid, n, spt, userDuration)
 				if err != nil {
 					return false, xerrors.Errorf("inserting into sectors_sdr_pipeline: %w", err)
 				}

cmd/curio/tasks/tasks.go

@@ -199,7 +199,8 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 		cfg.Subsystems.EnableBatchSeal ||
 		cfg.Subsystems.EnableUpdateEncode ||
 		cfg.Subsystems.EnableUpdateProve ||
-		cfg.Subsystems.EnableUpdateSubmit
+		cfg.Subsystems.EnableUpdateSubmit ||
+		cfg.Subsystems.EnableCommP
 
 	if hasAnySealingTask {
 		sealingTasks, err := addSealingTasks(ctx, hasAnySealingTask, db, full, sender, as, cfg, slrLazy, asyncParams, si, stor, bstore, machine)
@@ -237,7 +238,10 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 			// PSD and Deal find task do not require many resources. They can run on all machines
 			psdTask := storage_market.NewPSDTask(dm, db, sender, as, &cfg.Market.StorageMarketConfig.MK12, full)
 			dealFindTask := storage_market.NewFindDealTask(dm, db, full, &cfg.Market.StorageMarketConfig.MK12)
-			activeTasks = append(activeTasks, psdTask, dealFindTask)
+
+			checkIndexesTask := indexing.NewCheckIndexesTask(db, iStore)
+
+			activeTasks = append(activeTasks, psdTask, dealFindTask, checkIndexesTask)
 
 			// Start libp2p hosts and handle streams
 			err = libp2p.NewDealProvider(ctx, db, cfg, dm.MK12Handler, full, sender, miners, machine)
@@ -250,7 +254,7 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 
 		indexingTask := indexing.NewIndexingTask(db, sc, iStore, pp, cfg, idxMax)
 		ipniTask := indexing.NewIPNITask(db, sc, iStore, pp, cfg, idxMax)
-		activeTasks = append(activeTasks, indexingTask, ipniTask)
+		activeTasks = append(activeTasks, ipniTask, indexingTask)
 
 		if cfg.HTTP.Enable {
 			err = cuhttp.StartHTTPServer(ctx, dependencies)
@@ -277,6 +281,8 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 	}
 	go machineDetails(dependencies, activeTasks, ht.ResourcesAvailable().MachineID, dependencies.Name)
 
+	*dependencies.MachineID = int64(ht.ResourcesAvailable().MachineID)
+
 	if hasAnySealingTask {
 		watcher, err := message.NewMessageWatcher(db, ht, chainSched, full)
 		if err != nil {

cmd/curio/test-cli.go

@@ -28,6 +28,7 @@ var testCmd = &cli.Command{
 	Subcommands: []*cli.Command{
 		//provingInfoCmd,
 		wdPostCmd,
+		testDebugCmd,
 	},
 	Before: func(cctx *cli.Context) error {
 		return nil

cmd/curio/test-debug.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/filecoin-project/curio/deps"
+	"github.com/filecoin-project/curio/harmony/harmonydb"
+	"github.com/filecoin-project/curio/lib/reqcontext"
+	"github.com/filecoin-project/go-state-types/builtin/v9/market"
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+)
+
+var testDebugCmd = &cli.Command{
+	Name:  "debug",
+	Usage: "Collection of debugging utilities",
+	Subcommands: []*cli.Command{
+		testDebugIpniChunks,
+		testDebugMigPcid,
+	},
+}
+
+var testDebugMigPcid = &cli.Command{
+	Name: "mig-pcid",
+	Action: func(cctx *cli.Context) error {
+		ctx := reqcontext.ReqContext(cctx)
+		dep, err := deps.GetDepsCLI(ctx, cctx)
+		if err != nil {
+			return err
+		}
+
+		_, err = dep.DB.BeginTransaction(ctx, func(tx *harmonydb.Tx) (commit bool, err error) {
+			var bdeals []struct {
+				Prop json.RawMessage `db:"proposal"`
+				UUID string          `db:"uuid"`
+			}
+
+			err = tx.Select(&bdeals, `SELECT 
+										uuid,
+										proposal
+									FROM 
+										market_mk12_deals`)
+			if err != nil {
+				return false, xerrors.Errorf("getting deals from db: %w", err)
+			}
+
+			for _, d := range bdeals {
+				var prop market.DealProposal
+				err = json.Unmarshal(d.Prop, &prop)
+				if err != nil {
+					return false, xerrors.Errorf("unmarshal proposal: %w", err)
+				}
+
+				pcid, err := prop.Cid()
+				if err != nil {
+					return false, xerrors.Errorf("get cid: %w", err)
+				}
+
+				fmt.Println(d.UUID, pcid)
+
+				n, err := tx.Exec(`UPDATE market_mk12_deals SET proposal_cid = $1 WHERE uuid = $2`, pcid, d.UUID)
+				if err != nil {
+					return false, xerrors.Errorf("update deals: %w", err)
+				}
+				if n == 0 {
+					return false, xerrors.Errorf("update deals: deal not found")
+				}
+			}
+			return true, nil
+		})
+
+		return err
+	},
+}

cuhttp/server.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/CAFxX/httpcompression"
@@ -73,7 +74,7 @@ func compressionMiddleware(config *config.CompressionConfig) (func(http.Handler)
 func libp2pConnMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		// Check if the request path is "/"
-		if r.URL.Path == "/" {
+		if r.URL.Path == "/" || r.URL.Path == "" {
 			// Check if the request is a WebSocket upgrade request
 			if isWebSocketUpgrade(r) {
 				// Rewrite the path to "/libp2p"
@@ -92,26 +93,18 @@ func isWebSocketUpgrade(r *http.Request) bool {
 	if r.Method != http.MethodGet {
 		return false
 	}
-	if r.Header.Get("Upgrade") != "websocket" {
+	if strings.ToLower(r.Header.Get("Upgrade")) != "websocket" {
 		return false
 	}
-	if r.Header.Get("Connection") != "Upgrade" {
+	if strings.ToLower(r.Header.Get("Connection")) != "upgrade" {
 		return false
 	}
 	return true
 }
 
 func StartHTTPServer(ctx context.Context, d *deps.Deps) error {
-	ch := cache{db: d.DB}
 	cfg := d.Cfg.HTTP
 
-	// Set up the autocert manager for Let's Encrypt
-	certManager := autocert.Manager{
-		Cache:      ch,
-		Prompt:     autocert.AcceptTOS, // Automatically accept the Terms of Service
-		HostPolicy: autocert.HostWhitelist(cfg.DomainName),
-	}
-
 	// Setup the Chi router for more complex routing (if needed in the future)
 	chiRouter := chi.NewRouter()
 
@@ -160,20 +153,35 @@ func StartHTTPServer(ctx context.Context, d *deps.Deps) error {
 		Addr:              cfg.ListenAddress,
 		Handler:           libp2pConnMiddleware(loggingMiddleware(compressionMw(chiRouter))), // Attach middlewares
 		ReadTimeout:       cfg.ReadTimeout,
-		WriteTimeout:      cfg.WriteTimeout,
+		WriteTimeout:      time.Hour * 2,
 		IdleTimeout:       cfg.IdleTimeout,
 		ReadHeaderTimeout: cfg.ReadHeaderTimeout,
-		TLSConfig: &tls.Config{
+	}
+
+	if !cfg.DelegateTLS {
+		// Set up the autocert manager for Let's Encrypt
+		certManager := autocert.Manager{
+			Cache:      cache{db: d.DB},
+			Prompt:     autocert.AcceptTOS, // Automatically accept the Terms of Service
+			HostPolicy: autocert.HostWhitelist(cfg.DomainName),
+		}
+
+		server.TLSConfig = &tls.Config{
 			GetCertificate: certManager.GetCertificate,
-		},
+		}
 	}
 
 	// We don't need to run an HTTP server. Any HTTP request should simply be handled as HTTPS.
 
 	// Start the server with TLS
 	go func() {
 		log.Infof("Starting HTTPS server for https://%s on %s", cfg.DomainName, cfg.ListenAddress)
-		serr := server.ListenAndServeTLS("", "")
+		var serr error
+		if !cfg.DelegateTLS {
+			serr = server.ListenAndServeTLS("", "")
+		} else {
+			serr = server.ListenAndServe()
+		}
 		if serr != nil {
 			log.Errorf("Failed to start HTTPS server: %s", serr)
 			panic(serr)

deps/config/types.go

@@ -76,8 +76,8 @@ func DefaultCurioConfig() *CurioConfig {
 			StorageMarketConfig: StorageMarketConfig{
 				PieceLocator: []PieceLocatorConfig{},
 				Indexing: IndexingConfig{
-					InsertConcurrency: 8,
-					InsertBatchSize:   15000,
+					InsertConcurrency: 10,
+					InsertBatchSize:   1000,
 				},
 				MK12: MK12Config{
 					PublishMsgPeriod:          Duration(5 * time.Minute),
@@ -89,15 +89,13 @@ func DefaultCurioConfig() *CurioConfig {
 				IPNI: IPNIConfig{
 					ServiceURL:         []string{"https://cid.contact"},
 					DirectAnnounceURLs: []string{"https://cid.contact/ingest/announce"},
-					AnnounceAddresses:  []string{},
 				},
 			},
 		},
 		HTTP: HTTPConfig{
 			DomainName:        "",
 			ListenAddress:     "0.0.0.0:12310",
 			ReadTimeout:       time.Second * 10,
-			WriteTimeout:      time.Second * 10,
 			IdleTimeout:       time.Minute * 2,
 			ReadHeaderTimeout: time.Second * 5,
 			EnableCORS:        true,
@@ -588,10 +586,10 @@ type MK12Config struct {
 	// DisabledMiners is a list of miner addresses that should be excluded from online deal making protocols
 	DisabledMiners []string
 
-	// MaxConcurrentDealSize is a sum of all size of all deals which are waiting to be added to a sector
+	// MaxConcurrentDealSizeGiB is a sum of all size of all deals which are waiting to be added to a sector
 	// When the cumulative size of all deals in process reaches this number, new deals will be rejected.
 	// (Default: 0 = unlimited)
-	MaxConcurrentDealSize int64
+	MaxConcurrentDealSizeGiB int64
 
 	// DenyUnknownClients determines the default behaviour for the deal of clients which are not in allow/deny list
 	// If True then all deals coming from unknown clients will be rejected.
@@ -623,11 +621,6 @@ type IPNIConfig struct {
 	// The list of URLs of indexing nodes to announce to. This is a list of hosts we talk to tell them about new
 	// heads.
 	DirectAnnounceURLs []string
-
-	// AnnounceAddresses is a list of addresses indexer clients can use to reach to the HTTP market node.
-	// Curio allows running more than one node for HTTP server and thus all addressed can be announced
-	// simultaneously to the client. Example: ["https://mycurio.com", "http://myNewCurio:433/XYZ", "http://1.2.3.4:433"]
-	AnnounceAddresses []string
 }
 
 // HTTPConfig represents the configuration for an HTTP server.
@@ -642,12 +635,13 @@ type HTTPConfig struct {
 	// ListenAddress is the address that the server listens for HTTP requests.
 	ListenAddress string
 
+	// DelegateTLS allows the server to delegate TLS to a reverse proxy. When enabled the listen address will serve
+	// HTTP and the reverse proxy will handle TLS termination.
+	DelegateTLS bool
+
 	// ReadTimeout is the maximum duration for reading the entire or next request, including body, from the client.
 	ReadTimeout time.Duration
 
-	// WriteTimeout is the maximum duration before timing out writes of the response to the client.
-	WriteTimeout time.Duration
-
 	// IdleTimeout is the maximum duration of an idle session. If set, idle connections are closed after this duration.
 	IdleTimeout time.Duration