piece-directory: make sure we process recs correctly (#1828)

Author: nonsense

Makefile

@@ -217,7 +217,7 @@ docsgen-openrpc-boost: docsgen-openrpc-bin
 
 ## DOCKER IMAGES
 docker_user?=filecoin
-lotus_version?=v1.25.0-rc1
+lotus_version?=v1.25.0
 ffi_from_source?=0
 build_lotus?=0
 build_boost?=1

extern/boostd-data/yugabyte/service.go

@@ -550,6 +550,12 @@ func (s *Store) addMultihashesToPieces(ctx context.Context, pieceCid cid.Cid, re
 
 	threadBatch := len(recs) / s.settings.InsertConcurrency // split the slice into go-routine batches
 
+	if threadBatch == 0 {
+		threadBatch = len(recs)
+	}
+
+	log.Debugw("addMultihashesToPieces call", "threadBatch", threadBatch, "len(recs)", len(recs))
+
 	var eg errgroup.Group
 	for i := 0; i < len(recs); i += threadBatch {
 		i := i
@@ -576,9 +582,18 @@ func (s *Store) addMultihashesToPieces(ctx context.Context, pieceCid cid.Cid, re
 				})
 
 				if allIdx == len(recsb)-1 || len(batch.Entries) == s.settings.InsertBatchSize {
-					err := s.session.ExecuteBatch(batch)
+					err := func() error {
+						defer func(start time.Time) {
+							log.Debugw("addMultihashesToPieces executeBatch", "took", time.Since(start), "entries", len(batch.Entries))
+						}(time.Now())
+						err := s.session.ExecuteBatch(batch)
+						if err != nil {
+							return fmt.Errorf("inserting into PayloadToPieces: %w", err)
+						}
+						return nil
+					}()
 					if err != nil {
-						return fmt.Errorf("inserting into PayloadToPieces: %w", err)
+						return err
 					}
 					batch = nil
 
@@ -609,6 +624,12 @@ func (s *Store) addPieceInfos(ctx context.Context, pieceCid cid.Cid, recs []mode
 
 	threadBatch := len(recs) / s.settings.InsertConcurrency // split the slice into go-routine batches
 
+	if threadBatch == 0 {
+		threadBatch = len(recs)
+	}
+
+	log.Debugw("addPieceInfos call", "threadBatch", threadBatch, "len(recs)", len(recs))
+
 	var eg errgroup.Group
 	for i := 0; i < len(recs); i += threadBatch {
 		i := i
@@ -635,9 +656,19 @@ func (s *Store) addPieceInfos(ctx context.Context, pieceCid cid.Cid, recs []mode
 				})
 
 				if allIdx == len(recsb)-1 || len(batch.Entries) == s.settings.InsertBatchSize {
-					err := s.session.ExecuteBatch(batch)
+					err := func() error {
+						defer func(start time.Time) {
+							log.Debugw("addPieceInfos executeBatch", "took", time.Since(start), "entries", len(batch.Entries))
+						}(time.Now())
+
+						err := s.session.ExecuteBatch(batch)
+						if err != nil {
+							return fmt.Errorf("executing offset / size batch insert for piece %s: %w", pieceCid, err)
+						}
+						return nil
+					}()
 					if err != nil {
-						return fmt.Errorf("executing offset / size batch insert for piece %s: %w", pieceCid, err)
+						return err
 					}
 					batch = nil
 

go.mod

@@ -320,7 +320,7 @@ require (
 require (
 	github.com/filecoin-project/boost-gfm v1.26.7
 	github.com/filecoin-project/boost-graphsync v0.13.9
-	github.com/filecoin-project/boost/extern/boostd-data v0.0.0-20231101173716-1622d0ce2581
+	github.com/filecoin-project/boost/extern/boostd-data v0.0.0-20231124125934-3233c510357f
 	github.com/filecoin-project/go-data-transfer/v2 v2.0.0-rc7
 	github.com/filecoin-project/go-fil-markets v1.28.3
 	github.com/filecoin-project/lotus v1.25.0

go.sum

@@ -313,8 +313,8 @@ github.com/filecoin-project/boost-gfm v1.26.7 h1:ENJEqx1OzY072QnUP37YrGVmUiCewRw
 github.com/filecoin-project/boost-gfm v1.26.7/go.mod h1:OhG2y7WeDx3KU9DPjgWllS+3/ospPjm8/XDrvN6uOfk=
 github.com/filecoin-project/boost-graphsync v0.13.9 h1:RQepfTlffLGUmp3Ff7VosYrWUKPLiz++GGV2D/gIfuw=
 github.com/filecoin-project/boost-graphsync v0.13.9/go.mod h1:bc2M5ZLZJtXHl8kjnqtn4L1MsdEqpJErDaIeY0bJ9wk=
-github.com/filecoin-project/boost/extern/boostd-data v0.0.0-20231101173716-1622d0ce2581 h1:a2rhYL8QBTxmP1/E2VhfsvUB6CoCuS4/jEvSpOl68tU=
-github.com/filecoin-project/boost/extern/boostd-data v0.0.0-20231101173716-1622d0ce2581/go.mod h1:/d9yXj2CqhPPtM+m8GiH7rQHSGRqXYFg3V82Qsk34NA=
+github.com/filecoin-project/boost/extern/boostd-data v0.0.0-20231124125934-3233c510357f h1:8dd0yAadyeOL5Qd42XhEwD60UKvIFkY2MLhef/IaeOk=
+github.com/filecoin-project/boost/extern/boostd-data v0.0.0-20231124125934-3233c510357f/go.mod h1:MyzvfYWAH0OAyf95TLUWYq3cO3vm/TVzDS57GKQi47o=
 github.com/filecoin-project/dagstore v0.7.0 h1:IS0R+69za8dguYWeqz/MI+nb7ONpk03tAkxPCBXEKm0=
 github.com/filecoin-project/dagstore v0.7.0/go.mod h1:YKn4qXih+/2xQWpfJsaKGOi4POw5vH5grDmfPCCnx8g=
 github.com/filecoin-project/filecoin-ffi v0.30.4-0.20200910194244-f640612a1a1f/go.mod h1:+If3s2VxyjZn+KGGZIoRXBDSFQ9xL404JBJGf4WhEj0=

piecedirectory/piecedirectory.go

@@ -88,6 +88,12 @@ func NewPieceDirectory(store *bdclient.Store, pr types.PieceReader, addIndexThro
 		opt(pd.settings)
 	}
 
+	if pd.settings.addIndexConcurrency == 0 {
+		pd.settings.addIndexConcurrency = config.DefaultAddIndexConcurrency
+	}
+
+	log.Infow("new piece directory", "add-index-concurrency", pd.settings.addIndexConcurrency, "add-idx-throttle-size", pd.addIdxThrottleSize)
+
 	expireCallback := func(key string, reason ttlcache.EvictionReason, value interface{}) {
 		log.Debugw("expire callback", "piececid", key, "reason", reason)
 
@@ -317,6 +323,11 @@ func (ps *PieceDirectory) addIndexForPiece(ctx context.Context, pieceCid cid.Cid
 		return fmt.Errorf("generating index for piece %s: %w", pieceCid, err)
 	}
 
+	if len(recs) == 0 {
+		log.Warnw("add index: generated index with 0 recs", "pieceCid", pieceCid)
+		return nil
+	}
+
 	// Transferring a large number of records over the wire can take a significant amount of time.
 	// Split the transfer into multiple concurrent parts to speed it up. Note that the index can't be generated by boost-data itself
 	// as it doesn't possess the actual file.
@@ -339,7 +350,7 @@ func (ps *PieceDirectory) addIndexForPiece(ctx context.Context, pieceCid cid.Cid
 			}
 			// Add mh => piece index to store: "which piece contains the multihash?"
 			// Add mh => offset index to store: "what is the offset of the multihash within the piece?"
-			log.Debugw("add index: store index in local index directory", "pieceCid", pieceCid, "chunk", i, "chunksTotal", concurrency)
+			log.Debugw("add index: store index in local index directory", "pieceCid", pieceCid, "chunk", i, "chunksTotal", concurrency, "len(recs)", len(recs[start:end]), "start", start, "end", end)
 			if err := ps.store.AddIndex(ctx, pieceCid, recs[start:end], true); err != nil {
 				return fmt.Errorf("adding CAR index for piece %s: %w", pieceCid, err)
 			}