[WIP] feat: market 2.0

Dockerfile

@@ -14,6 +14,11 @@ ENV RUSTUP_HOME=/usr/local/rustup \
     PATH=/usr/local/cargo/bin:$PATH \
     RUST_VERSION=1.63.0
 
+COPY ./ /opt/curio
+WORKDIR /opt/curio
+RUN git submodule update --init
+RUN go mod download
+
 RUN set -eux; \
     dpkgArch="$(dpkg --print-architecture)"; \
     case "${dpkgArch##*-}" in \
@@ -32,9 +37,6 @@ RUN set -eux; \
     cargo --version; \
     rustc --version;
 
-COPY ./ /opt/curio
-WORKDIR /opt/curio
-
 ### make configurable filecoin-ffi build
 ARG FFI_BUILD_FROM_SOURCE=0
 ENV FFI_BUILD_FROM_SOURCE=${FFI_BUILD_FROM_SOURCE}

Makefile

@@ -266,7 +266,11 @@ go-generate:
 gen: gensimple
 .PHONY: gen
 
-gensimple: api-gen go-generate cfgdoc-gen docsgen docsgen-cli
+marketgen:
+	$(GOCC) run ./market/mk20/mk20gen -pkg ./market/mk20 -output ./market/mk20/http/info.md
+.PHONY: marketgen
+
+gensimple: api-gen go-generate cfgdoc-gen docsgen marketgen docsgen-cli
 	$(GOCC) run ./scripts/fiximports
 	go mod tidy
 .PHONY: gen

alertmanager/alerts.go

@@ -731,7 +731,7 @@ func missingSectorCheck(al *alerts) {
 		SectorID int64 `db:"sector_num"`
 	}
 
-	err := al.db.Select(al.ctx, &sectors, `SELECT miner_id, sector_num  FROM sector_location WHERE sector_filetype = 2 GROUP BY miner_id, sector_num ORDER BY miner_id, sector_num`)
+	err := al.db.Select(al.ctx, &sectors, `SELECT miner_id, sector_num  FROM sector_location WHERE sector_filetype = ANY(ARRAY[2,8]) GROUP BY miner_id, sector_num ORDER BY miner_id, sector_num`)
 	if err != nil {
 		al.alertMap[Name].err = xerrors.Errorf("getting sealed sectors from database: %w", err)
 		return

alertmanager/plugin/slack_webhook.go

@@ -65,29 +65,60 @@ func (s *SlackWebhook) SendAlert(data *AlertPayload) error {
 
 	// Iterate through the map to construct the remaining blocks
 	for key, value := range data.Details {
-		// Split value into sentences by period followed by space
+		// Split value into sentences by period followed by space.
 		sentences := strings.Split(value.(string), ". ")
-		formattedValue := fmt.Sprintf("• *%s*\n", key)
 
-		// Add a bullet point before each trimmed sentence
+		// Add the key as the header for each block.
+		baseFormattedValue := fmt.Sprintf("• *%s*\n", key)
+		currentFormattedValue := baseFormattedValue
+
+		// Keep track of the character limit (3000) when adding sentences.
 		for _, sentence := range sentences {
-			trimmedSentence := strings.TrimSpace(sentence) // Trim leading and trailing spaces
+			trimmedSentence := strings.TrimSpace(sentence) // Trim leading and trailing spaces.
 			if trimmedSentence != "" {
-				formattedValue += fmt.Sprintf("• %s.\n", trimmedSentence) // Add period back and newline
+				// Add a bullet point and sentence, restoring the period and newline.
+				newSection := fmt.Sprintf("• %s.\n", trimmedSentence)
+
+				// Check if adding this section exceeds the 3000-character limit.
+				if len(currentFormattedValue)+len(newSection) > 3000 {
+					// If limit exceeds, add the currentFormattedValue block to payload and start a new block.
+					payload.Blocks = append(payload.Blocks,
+						Block{
+							Type: "section",
+							Text: &TextBlock{
+								Type: "mrkdwn",
+								Text: currentFormattedValue,
+							},
+						},
+						Block{
+							Type: "divider",
+						},
+					)
+
+					// Start a new formatted value with the baseFormattedValue.
+					currentFormattedValue = baseFormattedValue
+				}
+
+				// Append the newSection to the currentFormattedValue.
+				currentFormattedValue += newSection
 			}
 		}
-		payload.Blocks = append(payload.Blocks,
-			Block{
-				Type: "section",
-				Text: &TextBlock{
-					Type: "mrkdwn",
-					Text: formattedValue,
+
+		// Add the last block if it contains any content.
+		if currentFormattedValue != baseFormattedValue {
+			payload.Blocks = append(payload.Blocks,
+				Block{
+					Type: "section",
+					Text: &TextBlock{
+						Type: "mrkdwn",
+						Text: currentFormattedValue,
+					},
 				},
-			},
-			Block{
-				Type: "divider",
-			},
-		)
+				Block{
+					Type: "divider",
+				},
+			)
+		}
 	}
 
 	// Marshal the payload to JSON
@@ -163,7 +194,8 @@ func (s *SlackWebhook) SendAlert(data *AlertPayload) error {
 			}
 		})
 	if err != nil {
-		return fmt.Errorf("after %d retries,last error: %w", iter, err)
+		log.Errorw("Slack Webhook payload:", string(jsonData))
+		return fmt.Errorf("after %d retries,last error: %w, %s", iter, err, string(jsonData))
 	}
 	return nil
 }

cmd/curio/tasks/tasks.go

@@ -99,7 +99,6 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 	machine := dependencies.ListenAddr
 	prover := dependencies.Prover
 	iStore := dependencies.IndexStore
-	pp := dependencies.SectorReader
 
 	var activeTasks []harmonytask.TaskInterface
 
@@ -226,7 +225,8 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 				return nil, err
 			}
 			cleanupPieceTask := piece2.NewCleanupPieceTask(db, must.One(slrLazy.Val()), 0)
-			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask)
+			aggregateChunksTask := piece2.NewAggregateChunksTask(db, lstor, stor)
+			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask, aggregateChunksTask)
 		}
 	}
 
@@ -244,21 +244,27 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 	}
 
 	{
+		var sdeps cuhttp.ServiceDeps
 		// Market tasks
 		var dm *storage_market.CurioStorageDealMarket
 		if cfg.Subsystems.EnableDealMarket {
 			// Main market poller should run on all nodes
-			dm = storage_market.NewCurioStorageDealMarket(miners, db, cfg, si, full, as)
+			dm = storage_market.NewCurioStorageDealMarket(miners, db, cfg, must.One(dependencies.EthClient.Val()), si, full, as, lstor)
 			err := dm.StartMarket(ctx)
 			if err != nil {
 				return nil, err
 			}
 
+			sdeps.DealMarket = dm
+
 			if cfg.Subsystems.EnableCommP {
 				commpTask := storage_market.NewCommpTask(dm, db, must.One(slrLazy.Val()), full, cfg.Subsystems.CommPMaxTasks)
 				activeTasks = append(activeTasks, commpTask)
 			}
 
+			aggTask := storage_market.NewAggregateTask(dm, db, must.One(slrLazy.Val()), lstor, full)
+			activeTasks = append(activeTasks, aggTask)
+
 			// 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)
@@ -275,7 +281,6 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 		if err != nil {
 			return nil, err
 		}
-		var sdeps cuhttp.ServiceDeps
 
 		if cfg.Subsystems.EnablePDP {
 			es := getSenderEth()
@@ -293,12 +298,12 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 
 		idxMax := taskhelp.Max(cfg.Subsystems.IndexingMaxTasks)
 
-		indexingTask := indexing.NewIndexingTask(db, sc, iStore, pp, cfg, idxMax)
-		ipniTask := indexing.NewIPNITask(db, sc, iStore, pp, cfg, idxMax)
+		indexingTask := indexing.NewIndexingTask(db, sc, iStore, dependencies.SectorReader, dependencies.CachedPieceReader, cfg, idxMax)
+		ipniTask := indexing.NewIPNITask(db, sc, iStore, dependencies.SectorReader, dependencies.CachedPieceReader, cfg, idxMax)
 		activeTasks = append(activeTasks, ipniTask, indexingTask)
 
 		if cfg.HTTP.Enable {
-			err = cuhttp.StartHTTPServer(ctx, dependencies, &sdeps, dm)
+			err = cuhttp.StartHTTPServer(ctx, dependencies, &sdeps)
 			if err != nil {
 				return nil, xerrors.Errorf("failed to start the HTTP server: %w", err)
 			}

cmd/pdptool/main.go

@@ -195,6 +195,8 @@ var pingCmd = &cli.Command{
 			return errCreateToken
 		}
 
+		serviceURL = serviceURL + "/market"
+
 		// Append /pdp/ping to the service URL
 		pingURL := serviceURL + "/pdp/ping"
 
@@ -512,6 +514,7 @@ var pieceUploadCmd = &cli.Command{
 		}
 
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		jwtToken := cctx.String("jwt-token")
 		notifyURL := cctx.String("notify-url")
 		serviceName := cctx.String("service-name")
@@ -669,6 +672,7 @@ var uploadFileCmd = &cli.Command{
 		}
 
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		jwtToken := cctx.String("jwt-token")
 		serviceName := cctx.String("service-name")
 		hashType := cctx.String("hash-type")
@@ -866,6 +870,7 @@ var createProofSetCmd = &cli.Command{
 	},
 	Action: func(cctx *cli.Context) error {
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		serviceName := cctx.String("service-name")
 		recordKeeper := cctx.String("recordkeeper")
 		extraDataHexStr := cctx.String("extra-data")
@@ -961,6 +966,7 @@ var getProofSetStatusCmd = &cli.Command{
 	},
 	Action: func(cctx *cli.Context) error {
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		serviceName := cctx.String("service-name")
 		txHash := cctx.String("tx-hash")
 
@@ -1072,6 +1078,7 @@ var getProofSetCmd = &cli.Command{
 		}
 
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		serviceName := cctx.String("service-name")
 
 		// Load the private key
@@ -1178,6 +1185,7 @@ var addRootsCmd = &cli.Command{
 	},
 	Action: func(cctx *cli.Context) error {
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		serviceName := cctx.String("service-name")
 		proofSetID := cctx.Uint64("proof-set-id")
 		rootInputs := cctx.StringSlice("root")
@@ -1422,6 +1430,7 @@ var removeRootsCmd = &cli.Command{
 	},
 	Action: func(cctx *cli.Context) error {
 		serviceURL := cctx.String("service-url")
+		serviceURL = serviceURL + "/market"
 		serviceName := cctx.String("service-name")
 		proofSetID := cctx.Uint64("proof-set-id")
 		rootID := cctx.Uint64("root-id")

cmd/sptool/main.go

@@ -101,5 +101,6 @@ var toolboxCmd = &cli.Command{
 	Subcommands: []*cli.Command{
 		sparkCmd,
 		mk12Clientcmd,
+		mk20Clientcmd,
 	},
 }