Feat/tssdk

.github/workflows/ci.yml

@@ -179,6 +179,8 @@ jobs:
             target: "./itests/harmonydb_test.go"
           - name: test-itest-alertnow
             target: "./itests/alertnow_test.go"
+          - name: test-itest-pdp-prove
+            target: "./itests/pdp_prove_test.go"
     steps:
       - uses: actions/checkout@v4
 
@@ -311,6 +313,10 @@ jobs:
         run: go install github.com/hannahhoward/cbor-gen-for
         shell: bash
 
+      - name: Install swag cli
+        run: go install github.com/swaggo/swag/cmd/[email protected]
+        shell: bash
+
 #      - name: Install gotext
 #        run: go install golang.org/x/text/cmd/gotext
 #        shell: bash

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}
@@ -56,13 +58,26 @@ RUN go install github.com/ipld/go-car/cmd/car@latest \
 RUN go install github.com/LexLuthr/piece-server@latest \
  && cp $GOPATH/bin/piece-server /usr/local/bin/
 
-RUN go install github.com/ipni/storetheindex@v0.8.38 \
+RUN go install github.com/ipni/storetheindex@latest \
  && cp $GOPATH/bin/storetheindex /usr/local/bin/
 
+RUN go install github.com/ethereum/go-ethereum/cmd/geth@latest \
+ && cp $GOPATH/bin/geth /usr/local/bin/
+
 #####################################
 FROM ubuntu:22.04 AS curio-all-in-one
 
-RUN apt-get update && apt-get install -y dnsutils vim curl aria2 jq
+RUN apt-get update && apt-get install -y dnsutils vim curl aria2 jq git wget nodejs npm
+
+# Install Foundry
+RUN curl -L https://foundry.paradigm.xyz | bash \
+    && bash -c ". ~/.foundry/bin/foundryup"
+
+# Make sure foundry binaries are available in PATH
+ENV PATH="/root/.foundry/bin:${PATH}"
+
+# Verify installation
+RUN forge --version && cast --version && anvil --version
 
 # Copy libraries and binaries from curio-builder
 COPY --from=curio-builder /etc/ssl/certs /etc/ssl/certs
@@ -98,6 +113,7 @@ COPY --from=curio-builder /opt/curio/sptool /usr/local/bin/
 COPY --from=piece-server-builder /usr/local/bin/piece-server /usr/local/bin/
 COPY --from=piece-server-builder /usr/local/bin/car /usr/local/bin/
 COPY --from=piece-server-builder /usr/local/bin/storetheindex /usr/local/bin/
+COPY --from=piece-server-builder /usr/local/bin/geth /usr/local/bin/
 
 # Set up directories and permissions
 RUN mkdir /var/tmp/filecoin-proof-parameters \

Makefile

@@ -266,7 +266,11 @@ go-generate:
 gen: gensimple
 .PHONY: gen
 
-gensimple: api-gen go-generate cfgdoc-gen docsgen docsgen-cli
+marketgen:
+	swag init -dir market/mk20/http -g http.go  -o market/mk20/http --parseDependencyLevel 3 --parseDependency
+.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

@@ -3,7 +3,6 @@ package alertmanager
 import (
 	"bytes"
 	"context"
-	"database/sql"
 	"fmt"
 	"math"
 	"strings"
@@ -13,6 +12,7 @@ import (
 	"github.com/dustin/go-humanize"
 	cbor "github.com/ipfs/go-ipld-cbor"
 	"github.com/samber/lo"
+	"github.com/yugabyte/pgx/v5"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
@@ -346,7 +346,7 @@ func (al *alerts) getAddresses() ([]address.Address, []address.Address, error) {
 		cfg := config.DefaultCurioConfig()
 		err := al.db.QueryRow(al.ctx, `SELECT config FROM harmony_config WHERE title=$1`, layer).Scan(&text)
 		if err != nil {
-			if strings.Contains(err.Error(), sql.ErrNoRows.Error()) {
+			if strings.Contains(err.Error(), pgx.ErrNoRows.Error()) {
 				return nil, nil, xerrors.Errorf("missing layer '%s' ", layer)
 			}
 			return nil, nil, xerrors.Errorf("could not read layer '%s': %w", layer, err)
@@ -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

@@ -102,7 +102,6 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 	machine := dependencies.ListenAddr
 	prover := dependencies.Prover
 	iStore := dependencies.IndexStore
-	pp := dependencies.SectorReader
 
 	chainSched := chainsched.New(full)
 
@@ -234,12 +233,13 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 	{
 		// Piece handling
 		if cfg.Subsystems.EnableParkPiece {
-			parkPieceTask, err := piece2.NewParkPieceTask(db, must.One(slrLazy.Val()), cfg.Subsystems.ParkPieceMaxTasks)
+			parkPieceTask, err := piece2.NewParkPieceTask(db, must.One(slrLazy.Val()), stor, cfg.Subsystems.ParkPieceMaxTasks)
 			if err != nil {
 				return nil, err
 			}
 			cleanupPieceTask := piece2.NewCleanupPieceTask(db, must.One(slrLazy.Val()), 0)
-			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask)
+			aggregateChunksTask := piece2.NewAggregateChunksTask(db, stor, must.One(slrLazy.Val()))
+			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask, aggregateChunksTask)
 		}
 	}
 
@@ -257,21 +257,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, must.One(slrLazy.Val()))
 			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)
@@ -288,30 +294,48 @@ 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()
 			sdeps.EthSender = es
 
-			pdp.NewWatcherCreate(db, must.One(dependencies.EthClient.Val()), chainSched)
-			pdp.NewWatcherRootAdd(db, must.One(dependencies.EthClient.Val()), chainSched)
+			ethClient := must.One(dependencies.EthClient.Val())
+
+			pdp.NewWatcherDataSetCreate(db, ethClient, chainSched)
+			pdp.NewWatcherPieceAdd(db, chainSched, ethClient)
+			pdp.NewWatcherDelete(db, chainSched)
+			pdp.NewWatcherPieceDelete(db, chainSched)
 
-			pdpProveTask := pdp.NewProveTask(chainSched, db, must.One(dependencies.EthClient.Val()), dependencies.Chain, es, dependencies.CachedPieceReader)
-			pdpNextProvingPeriodTask := pdp.NewNextProvingPeriodTask(db, must.One(dependencies.EthClient.Val()), dependencies.Chain, chainSched, es)
-			pdpInitProvingPeriodTask := pdp.NewInitProvingPeriodTask(db, must.One(dependencies.EthClient.Val()), dependencies.Chain, chainSched, es)
+			pdpProveTask := pdp.NewProveTask(chainSched, db, ethClient, dependencies.Chain, es, dependencies.CachedPieceReader, iStore)
+			pdpNextProvingPeriodTask := pdp.NewNextProvingPeriodTask(db, ethClient, dependencies.Chain, chainSched, es)
+			pdpInitProvingPeriodTask := pdp.NewInitProvingPeriodTask(db, ethClient, dependencies.Chain, chainSched, es)
 			pdpNotifTask := pdp.NewPDPNotifyTask(db)
-			activeTasks = append(activeTasks, pdpNotifTask, pdpProveTask, pdpNextProvingPeriodTask, pdpInitProvingPeriodTask)
+
+			addProofSetTask := pdp.NewPDPTaskAddDataSet(db, es, ethClient, full)
+			pdpAddRoot := pdp.NewPDPTaskAddPiece(db, es, ethClient)
+			pdpDelRoot := pdp.NewPDPTaskDeletePiece(db, es, ethClient)
+			pdpDelProofSetTask := pdp.NewPDPTaskDeleteDataSet(db, es, ethClient, full)
+
+			pdpAggregateTask := pdp.NewAggregatePDPDealTask(db, sc)
+			pdpCache := pdp.NewTaskPDPSaveCache(db, dependencies.CachedPieceReader, iStore)
+			commPTask := pdp.NewPDPCommpTask(db, sc, cfg.Subsystems.CommPMaxTasks)
+
+			activeTasks = append(activeTasks, pdpNotifTask, pdpProveTask, pdpNextProvingPeriodTask, pdpInitProvingPeriodTask, commPTask, pdpAddRoot, addProofSetTask, pdpAggregateTask, pdpCache, pdpDelRoot, pdpDelProofSetTask)
 		}
 
 		idxMax := taskhelp.Max(cfg.Subsystems.IndexingMaxTasks)
 
-		indexingTask := indexing.NewIndexingTask(db, sc, iStore, pp, cfg, idxMax)
-		ipniTask := indexing.NewIPNITask(db, sc, iStore, pp, cfg, idxMax)
-		activeTasks = append(activeTasks, ipniTask, indexingTask)
+		indexingTask := indexing.NewIndexingTask(db, sc, iStore, dependencies.SectorReader, dependencies.CachedPieceReader, cfg, idxMax)
+		ipniTask := indexing.NewIPNITask(db, sc, dependencies.SectorReader, dependencies.CachedPieceReader, cfg, idxMax)
+		pdpIdxTask := indexing.NewPDPIndexingTask(db, sc, iStore, dependencies.CachedPieceReader, cfg, idxMax)
+		pdpIPNITask := indexing.NewPDPIPNITask(db, sc, dependencies.CachedPieceReader, cfg, idxMax)
+		activeTasks = append(activeTasks, ipniTask, indexingTask, pdpIdxTask, pdpIPNITask)
 
 		if cfg.HTTP.Enable {
-			err = cuhttp.StartHTTPServer(ctx, dependencies, &sdeps, dm)
+			if !cfg.Subsystems.EnableDealMarket {
+				return nil, xerrors.New("deal market must be enabled on HTTP server")
+			}
+			err = cuhttp.StartHTTPServer(ctx, dependencies, &sdeps)
 			if err != nil {
 				return nil, xerrors.Errorf("failed to start the HTTP server: %w", err)
 			}
@@ -321,6 +345,9 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 	amTask := alertmanager.NewAlertTask(full, db, cfg.Alerting, dependencies.Al)
 	activeTasks = append(activeTasks, amTask)
 
+	pcl := gc.NewPieceCleanupTask(db, iStore)
+	activeTasks = append(activeTasks, pcl)
+
 	log.Infow("This Curio instance handles",
 		"miner_addresses", miners,
 		"tasks", lo.Map(activeTasks, func(t harmonytask.TaskInterface, _ int) string { return t.TypeDetails().Name }))