filecoin-project
diff --git a/‎cmd/curio/debug-ipni.go
+55 b/‎cmd/curio/debug-ipni.go
+55
diff --git a/‎cmd/curio/seal.go
+1-1 b/‎cmd/curio/seal.go
+1-1
diff --git a/‎cmd/curio/tasks/tasks.go
+21-15 b/‎cmd/curio/tasks/tasks.go
+21-15
diff --git a/‎cmd/curio/test-cli.go
+1 b/‎cmd/curio/test-cli.go
+1
diff --git a/‎cmd/curio/test-debug.go
+13 b/‎cmd/curio/test-debug.go
+13
diff --git a/‎cuhttp/server.go
+23-15 b/‎cuhttp/server.go
+23-15
diff --git a/‎deps/config/doc_gen.go
+44-23 b/‎deps/config/doc_gen.go
+44-23
@@ -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
+	},
+}
@@ -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)
 				}
 
@@ -178,18 +178,6 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 		return ffi.NewSealCalls(stor, lstor, si), nil
 	})
 
-	{
-		// Piece handling
-		if cfg.Subsystems.EnableParkPiece {
-			parkPieceTask, err := piece2.NewParkPieceTask(db, must.One(slrLazy.Val()), cfg.Subsystems.ParkPieceMaxTasks)
-			if err != nil {
-				return nil, err
-			}
-			cleanupPieceTask := piece2.NewCleanupPieceTask(db, must.One(slrLazy.Val()), 0)
-			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask)
-		}
-	}
-
 	hasAnySealingTask := cfg.Subsystems.EnableSealSDR ||
 		cfg.Subsystems.EnableSealSDRTrees ||
 		cfg.Subsystems.EnableSendPrecommitMsg ||
@@ -199,7 +187,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)
@@ -209,6 +198,18 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 		activeTasks = append(activeTasks, sealingTasks...)
 	}
 
+	{
+		// Piece handling
+		if cfg.Subsystems.EnableParkPiece {
+			parkPieceTask, err := piece2.NewParkPieceTask(db, must.One(slrLazy.Val()), cfg.Subsystems.ParkPieceMaxTasks)
+			if err != nil {
+				return nil, err
+			}
+			cleanupPieceTask := piece2.NewCleanupPieceTask(db, must.One(slrLazy.Val()), 0)
+			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask)
+		}
+	}
+
 	miners := make([]address.Address, 0, len(maddrs))
 	for k := range maddrs {
 		miners = append(miners, address.Address(k))
@@ -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 {
 
@@ -28,6 +28,7 @@ var testCmd = &cli.Command{
 	Subcommands: []*cli.Command{
 		//provingInfoCmd,
 		wdPostCmd,
+		testDebugCmd,
 	},
 	Before: func(cctx *cli.Context) error {
 		return nil
 
@@ -0,0 +1,13 @@
+package main
+
+import (
+	"github.com/urfave/cli/v2"
+)
+
+var testDebugCmd = &cli.Command{
+	Name:  "debug",
+	Usage: "Collection of debugging utilities",
+	Subcommands: []*cli.Command{
+		testDebugIpniChunks,
+	},
+}
@@ -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)
Original file line number	Diff line number	Diff line change
`@@ -152,7 +152,7 @@ var sealStartCmd = &cli.Command{`
`152`	`152`	`}`
`153`	`153`
`154`	`154`	`for _, n := range num {`
`155`		`- _, err := tx.Exec("insert into sectors_sdr_pipeline (sp_id, sector_number, reg_seal_proof) values ($1, $2, $3)", mid, n, spt)`
	`155`	`+ _, 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)`
`156`	`156`	`if err != nil {`
`157`	`157`	`return false, xerrors.Errorf("inserting into sectors_sdr_pipeline: %w", err)`
`158`	`158`	`}`