Skip to content

[WIP] feat: market 2.0 #508

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 15 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -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 \
Expand All @@ -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}
Expand Down
6 changes: 5 additions & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion alertmanager/alerts.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
66 changes: 49 additions & 17 deletions alertmanager/plugin/slack_webhook.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
}
19 changes: 12 additions & 7 deletions cmd/curio/tasks/tasks.go
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
}
}

Expand All @@ -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)
Expand All @@ -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()
Expand All @@ -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)
}
Expand Down
9 changes: 9 additions & 0 deletions cmd/pdptool/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,8 @@ var pingCmd = &cli.Command{
return errCreateToken
}

serviceURL = serviceURL + "/market"

// Append /pdp/ping to the service URL
pingURL := serviceURL + "/pdp/ping"

Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down
1 change: 1 addition & 0 deletions cmd/sptool/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,5 +101,6 @@ var toolboxCmd = &cli.Command{
Subcommands: []*cli.Command{
sparkCmd,
mk12Clientcmd,
mk20Clientcmd,
},
}
Loading