Merge pull request #547 from ipfs-force-community/fix/hang

fix: Separate coroutine generation index
ipfs-force-community · Dec 25, 2024 · 5ea4f3d · 5ea4f3d
2 parents f4bffcd + bda77b2
commit 5ea4f3d
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 10 deletions.
diff --git a/go.mod b/go.mod
@@ -367,7 +367,7 @@ require (
 )
 
 replace (
-	github.com/filecoin-project/dagstore => github.com/ipfs-force-community/dagstore v0.4.4-0.20231218095803-2d4efb48c393
+	github.com/filecoin-project/dagstore => github.com/ipfs-force-community/dagstore v0.4.4-0.20241209090026-ab2f6df47e8e
 	github.com/filecoin-project/filecoin-ffi => ./extern/filecoin-ffi
 	github.com/filecoin-project/go-fil-markets => github.com/ipfs-force-community/go-fil-markets v1.2.6-0.20230822060005-aee2cbae5b01
 	github.com/filecoin-project/go-jsonrpc => github.com/ipfs-force-community/go-jsonrpc v0.1.9

diff --git a/go.sum b/go.sum
@@ -1004,8 +1004,8 @@ github.com/influxdata/influxdb-client-go/v2 v2.2.2/go.mod h1:fa/d1lAdUHxuc1jedx3
 github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 h1:W9WBk7wlPfJLvMCdtV4zPulc4uCPrlywQOmbFOhgQNU=
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
-github.com/ipfs-force-community/dagstore v0.4.4-0.20231218095803-2d4efb48c393 h1:DZd7I0MLgFtW8sVeSk7MSEDLC8PQ4Drl01CIKak/JM0=
-github.com/ipfs-force-community/dagstore v0.4.4-0.20231218095803-2d4efb48c393/go.mod h1:YKn4qXih+/2xQWpfJsaKGOi4POw5vH5grDmfPCCnx8g=
+github.com/ipfs-force-community/dagstore v0.4.4-0.20241209090026-ab2f6df47e8e h1:vo0qbCobRa8pU5e+XZj0FhXwJVXLcdbc35OBG0dzh+Q=
+github.com/ipfs-force-community/dagstore v0.4.4-0.20241209090026-ab2f6df47e8e/go.mod h1:YKn4qXih+/2xQWpfJsaKGOi4POw5vH5grDmfPCCnx8g=
 github.com/ipfs-force-community/go-fil-markets v1.2.6-0.20230822060005-aee2cbae5b01 h1:sD9/GoGBui1j5fpKepe1v4P5NsjurqbM8WQlk4LNZeI=
 github.com/ipfs-force-community/go-fil-markets v1.2.6-0.20230822060005-aee2cbae5b01/go.mod h1:eryxo/oVgIxaR5g5CNr9PlvZOi+u/bak0IsPL/PT1hk=
 github.com/ipfs-force-community/go-jsonrpc v0.1.9 h1:5QavBltfvV6fz/+EbYsCkVxJ1MSJncZm6YuPs1SLdZU=

diff --git a/storageprovider/deal_handler.go b/storageprovider/deal_handler.go
@@ -544,12 +544,14 @@ func (storageDealPorcess *StorageDealProcessImpl) HandleOff(ctx context.Context,
 		}
 		log.Infof("after publishing deal. piece cid: %s, payload size: %d", deal.Proposal.PieceCID, deal.PayloadSize)
 
-		log.Infof("register shard. deal: %d, proposalCid: %s, pieceCid: %s", deal.DealID, deal.ProposalCid, deal.Proposal.PieceCID)
-		// Register the deal data as a "shard" with the DAG store. Later it can be
-		// fetched from the DAG store during retrieval.
-		if err := storageDealPorcess.dagStore.RegisterShard(ctx, deal.Proposal.PieceCID, carFilePath, true, nil); err != nil {
-			log.Errorf("failed to register shard: %v", err)
-		}
+		go func() {
+			log.Infof("register shard. deal: %d, proposalCid: %s, pieceCid: %s", deal.DealID, deal.ProposalCid, deal.Proposal.PieceCID)
+			// Register the deal data as a "shard" with the DAG store. Later it can be
+			// fetched from the DAG store during retrieval.
+			if err := storageDealPorcess.dagStore.RegisterShard(ctx, deal.Proposal.PieceCID, carFilePath, true, nil); err != nil {
+				log.Errorf("failed to register shard: %v", err)
+			}
+		}()
 
 		// Remove temporary car files
 		storageDealPorcess.removeTemporaryFile(ctx, deal, true)

diff --git a/storageprovider/storage_provider.go b/storageprovider/storage_provider.go
@@ -238,7 +238,7 @@ func (p *StorageProviderImpl) start(ctx context.Context) error {
 	// Run datastore and DAG store migrations
 	deals, err := p.dealStore.ListDeal(ctx, &types.StorageDealQueryParams{Page: types.Page{Limit: math.MaxInt32}})
 	if err != nil {
-		return nil
+		return fmt.Errorf("failed to list deals: %w", err)
 	}
 	// Fire restart event on all active deals
 	if err := p.restartDeals(ctx, deals); err != nil {
@@ -262,6 +262,7 @@ func (p *StorageProviderImpl) restartDeals(ctx context.Context, deals []*types.M
 			continue
 		}
 
+		log.Debugf("restart deal %s, state: %s, piece cid: %s", deal.ProposalCid, deal.State, deal.Proposal.PieceCID)
 		go func(deal *types.MinerDeal) {
 			err := p.dealProcess.HandleOff(ctx, deal)
 			if err != nil {