Merge pull request #8160 from filecoin-project/fix/addpiece-idempotent

feat: Make add piece idempotent

extern/storage-sealing/input.go

@@ -311,41 +311,68 @@ func (m *Sealing) SectorAddPieceToAny(ctx context.Context, size abi.UnpaddedPiec
 	}
 
 	m.inputLk.Lock()
-	if _, exist := m.pendingPieces[proposalCID(deal)]; exist {
+	if pp, exist := m.pendingPieces[proposalCID(deal)]; exist {
 		m.inputLk.Unlock()
-		return api.SectorOffset{}, xerrors.Errorf("piece for deal %s already pending", proposalCID(deal))
+
+		// we already have a pre-existing add piece call for this deal, let's wait for it to finish and see if it's successful
+		for {
+			res, err := waitAddPieceResp(ctx, pp)
+			if err != nil {
+				return api.SectorOffset{}, err
+			}
+			//  there was an error waiting for a pre-existing add piece call, let's retry
+			if res.err != nil {
+				m.inputLk.Lock()
+				pp = m.addPendingPiece(ctx, size, data, deal, sp)
+				m.inputLk.Unlock()
+				continue
+			}
+			// all good, return the response
+			return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
+		}
 	}
 
-	resCh := make(chan struct {
+	pp := m.addPendingPiece(ctx, size, data, deal, sp)
-		sn     abi.SectorNumber
+	m.inputLk.Unlock()
-		offset abi.UnpaddedPieceSize
+	res, err := waitAddPieceResp(ctx, pp)
-		err    error
+	if err != nil {
-	}, 1)
+		return api.SectorOffset{}, err
+	}
+	return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
+}
 
-	m.pendingPieces[proposalCID(deal)] = &pendingPiece{
+func (m *Sealing) addPendingPiece(ctx context.Context, size abi.UnpaddedPieceSize, data storage.Data, deal api.PieceDealInfo, sp abi.RegisteredSealProof) *pendingPiece {
+	doneCh := make(chan struct{})
+	pp := &pendingPiece{
+		doneCh:   doneCh,
 		size:     size,
 		deal:     deal,
 		data:     data,
 		assigned: false,
-		accepted: func(sn abi.SectorNumber, offset abi.UnpaddedPieceSize, err error) {
+	}
-			resCh <- struct {
+	pp.accepted = func(sn abi.SectorNumber, offset abi.UnpaddedPieceSize, err error) {
-				sn     abi.SectorNumber
+		pp.resp = &pieceAcceptResp{sn, offset, err}
-				offset abi.UnpaddedPieceSize
+		close(pp.doneCh)
-				err    error
-			}{sn: sn, offset: offset, err: err}
-		},
 	}
 
+	m.pendingPieces[proposalCID(deal)] = pp
 	go func() {
-		defer m.inputLk.Unlock()
 		if err := m.updateInput(ctx, sp); err != nil {
 			log.Errorf("%+v", err)
 		}
 	}()
 
-	res := <-resCh
+	return pp
+}
 
-	return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
+func waitAddPieceResp(ctx context.Context, pp *pendingPiece) (*pieceAcceptResp, error) {
+	select {
+	case <-pp.doneCh:
+		res := pp.resp
+		return res, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
 }
 
 func (m *Sealing) MatchPendingPiecesToOpenSectors(ctx context.Context) error {

extern/storage-sealing/sealing.go

@@ -140,7 +140,16 @@ func (o *openSector) dealFitsInLifetime(dealEnd abi.ChainEpoch, expF func(sn abi
 	return expiration >= dealEnd, nil
 }
 
+type pieceAcceptResp struct {
+	sn     abi.SectorNumber
+	offset abi.UnpaddedPieceSize
+	err    error
+}
+
 type pendingPiece struct {
+	doneCh chan struct{}
+	resp   *pieceAcceptResp
+
 	size abi.UnpaddedPieceSize
 	deal api.PieceDealInfo