Feat/allow small pieces

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.23.6-bullseye as builder
+FROM golang:1.23.6-bullseye AS builder
 WORKDIR /app
 COPY go.* ./
 RUN go mod download

cmd/api_test.go

@@ -263,7 +263,7 @@ func TestBasicDataPrep(t *testing.T) {
 		require.True(t, listPiecesResp.IsSuccess())
 		require.Len(t, listPiecesResp.Payload, 1)
 		require.Len(t, listPiecesResp.Payload[0].Pieces, 1)
-		require.Equal(t, "baga6ea4seaqoahdvfwkrp64ecsxbjvyuqcwpz3o7ctxrjanlv2x4u2cq2qjf2ji", listPiecesResp.Payload[0].Pieces[0].PieceCid)
+		require.Equal(t, "baga6ea4seaqhmks2wnochilik4updmit54agfi5mjf6r7ehotu36ksdp46uxahi", listPiecesResp.Payload[0].Pieces[0].PieceCid)
 		// Start daggen
 		startDagGenResp, err := client.Job.StartDagGen(&job.StartDagGenParams{
 			ID:      "prep",
@@ -285,7 +285,9 @@ func TestBasicDataPrep(t *testing.T) {
 		require.True(t, listPiecesResp.IsSuccess())
 		require.Len(t, listPiecesResp.Payload, 1)
 		require.Len(t, listPiecesResp.Payload[0].Pieces, 2)
-		require.Equal(t, "baga6ea4seaqoahdvfwkrp64ecsxbjvyuqcwpz3o7ctxrjanlv2x4u2cq2qjf2ji", listPiecesResp.Payload[0].Pieces[0].PieceCid)
-		require.Equal(t, "baga6ea4seaqbkouoyih2elxfrztq3gr23rpvgpx5e3fnud2rhvvzf4b7tneeyki", listPiecesResp.Payload[0].Pieces[1].PieceCid)
+		// data piece, full size
+		require.Equal(t, "baga6ea4seaqhmks2wnochilik4updmit54agfi5mjf6r7ehotu36ksdp46uxahi", listPiecesResp.Payload[0].Pieces[0].PieceCid)
+		// dag piece, min piece size
+		require.Equal(t, "baga6ea4seaqfoo2k3wmwp7gvxnc7hbjpb7ovtvt52tehwfvzxbreljcebbnwgiq", listPiecesResp.Payload[0].Pieces[1].PieceCid)
 	})
 }

cmd/dataprep/create.go

@@ -55,6 +55,12 @@ var CreateCmd = &cli.Command{
 			Value:       "",
 			DefaultText: "Determined by --max-size",
 		},
+		&cli.StringFlag{
+			Name:        "min-piece-size",
+			Usage:       "The minimum size of a piece. Pieces smaller than this will be padded up to this size. It's recommended to leave this as the default",
+			Value:       "1MiB",
+			DefaultText: "1MiB",
+		},
 		&cli.BoolFlag{
 			Name:  "delete-after-export",
 			Usage: "Whether to delete the source files after export to CAR files",
@@ -83,6 +89,7 @@ var CreateCmd = &cli.Command{
 		outputStorages := c.StringSlice("output")
 		maxSizeStr := c.String("max-size")
 		pieceSizeStr := c.String("piece-size")
+		minPieceSizeStr := c.String("min-piece-size")
 		for _, sourcePath := range c.StringSlice("local-source") {
 			source, err := createStorageIfNotExist(c.Context, db, sourcePath)
 			if err != nil {
@@ -103,8 +110,9 @@ var CreateCmd = &cli.Command{
 			OutputStorages:    outputStorages,
 			MaxSizeStr:        maxSizeStr,
 			PieceSizeStr:      pieceSizeStr,
-			DeleteAfterExport: c.Bool("delete-after-export"),
+			MinPieceSizeStr:   minPieceSizeStr,
 			Name:              name,
+			DeleteAfterExport: c.Bool("delete-after-export"),
 			NoInline:          c.Bool("no-inline"),
 			NoDag:             c.Bool("no-dag"),
 		})

cmd/functional_test.go

@@ -451,6 +451,10 @@ func TestDataPrep(t *testing.T) {
 							require.Equal(t, pieceCID, calculatedPieceCID)
 							err = os.WriteFile(filepath.Join(downloadDir, pieceCID+".car"), downloaded, 0777)
 							require.NoError(t, err)
+
+							// Verify piece size is a power of two
+							pieceSize := uint64(len(downloaded))
+							require.True(t, util.IsPowerOfTwo(pieceSize), "piece size %d is not a power of two", pieceSize)
 						}
 
 						// Download all pieces using local download server
@@ -499,7 +503,7 @@ func TestNoDuplicatedOutput(t *testing.T) {
 		_, _, err = runner.Run(ctx, fmt.Sprintf("singularity storage create local --name source --path %s", testutil.EscapePath(source)))
 		require.NoError(t, err)
 
-		_, _, err = runner.Run(ctx, fmt.Sprintf("singularity prep create --name test-prep --delete-after-export --source source --local-output %s --max-size=500KiB", testutil.EscapePath(output)))
+		_, _, err = runner.Run(ctx, fmt.Sprintf("singularity prep create --name test-prep --delete-after-export --source source --local-output %s --max-size=500KiB --min-piece-size=256KiB", testutil.EscapePath(output)))
 		require.NoError(t, err)
 
 		// Start scanning

handler/dataprep/create.go

@@ -18,6 +18,7 @@ type CreateRequest struct {
 	OutputStorages    []string `json:"outputStorages"`                          // Name of Output storage systems to be used for the output
 	MaxSizeStr        string   `default:"31.5GiB"     json:"maxSize"`           // Maximum size of the CAR files to be created
 	PieceSizeStr      string   `default:""            json:"pieceSize"`         // Target piece size of the CAR files used for piece commitment calculation
+	MinPieceSizeStr   string   `default:"1MiB"        json:"minPieceSize"`      // Minimum piece size for the preparation, applies only to DAG and remainer pieces
 	DeleteAfterExport bool     `default:"false"       json:"deleteAfterExport"` // Whether to delete the source files after export
 	NoInline          bool     `default:"false"       json:"noInline"`          // Whether to disable inline storage for the preparation. Can save database space but requires at least one output storage.
 	NoDag             bool     `default:"false"       json:"noDag"`             // Whether to disable maintaining folder dag structure for the sources. If disabled, DagGen will not be possible and folders will not have an associated CID.
@@ -77,6 +78,24 @@ func ValidateCreateRequest(ctx context.Context, db *gorm.DB, request CreateReque
 		return nil, errors.Wrap(handlererror.ErrInvalidParameter, "maxSize needs to be reduced to leave space for padding")
 	}
 
+	minPieceSizeStr := request.MinPieceSizeStr
+	if minPieceSizeStr == "" {
+		minPieceSizeStr = "1MiB"
+	}
+
+	minPieceSize, err := humanize.ParseBytes(minPieceSizeStr)
+	if err != nil {
+		return nil, errors.Join(handlererror.ErrInvalidParameter, errors.Wrapf(err, "invalid value for minPieceSize: %s", minPieceSizeStr))
+	}
+
+	if minPieceSize > pieceSize {
+		return nil, errors.Wrap(handlererror.ErrInvalidParameter, "minPieceSize cannot be larger than pieceSize")
+	}
+
+	if minPieceSize != util.NextPowerOfTwo(minPieceSize) {
+		return nil, errors.Wrap(handlererror.ErrInvalidParameter, "minPieceSize must be a power of two")
+	}
+
 	var sources []model.Storage
 	for _, name := range request.SourceStorages {
 		var source model.Storage
@@ -114,6 +133,7 @@ func ValidateCreateRequest(ctx context.Context, db *gorm.DB, request CreateReque
 	return &model.Preparation{
 		MaxSize:           int64(maxSize),
 		PieceSize:         int64(pieceSize),
+		MinPieceSize:      int64(minPieceSize),
 		SourceStorages:    sources,
 		OutputStorages:    outputs,
 		DeleteAfterExport: request.DeleteAfterExport,

handler/dataprep/piece.go

@@ -210,6 +210,7 @@ func (DefaultHandler) AddPieceHandler(
 		StoragePath:   request.FilePath,
 		PreparationID: preparation.ID,
 		FileSize:      fileSize,
+		PieceType:     model.DataPiece,
 	}
 
 	err = database.DoRetry(ctx, func() error { return db.Create(&mCar).Error })

handler/job/pack_test.go

@@ -137,7 +137,7 @@ func TestPackHandler_Success(t *testing.T) {
 		require.NoError(t, err)
 		require.NotNil(t, car)
 		require.EqualValues(t, 100, car.FileSize)
-		require.EqualValues(t, "baga6ea4seaqbuglmtahbspkbeunqohciieh4yjivfhcqawufwgs4gt7mzmyfmmi", car.PieceCID.String())
+		require.EqualValues(t, "baga6ea4seaqpikooah5wmbpjmnvx3ysyf36xagymjtbccnf5twt2cpaqcgcwqha", car.PieceCID.String())
 		err = db.Find(&job, 1).Error
 		require.NoError(t, err)
 		require.Equal(t, model.Complete, job.State)

model/preparation.go

@@ -11,6 +11,13 @@ import (
 	"gorm.io/gorm"
 )
 
+type PieceType string
+
+const (
+	DataPiece PieceType = "data"
+	DagPiece  PieceType = "dag"
+)
+
 type Worker struct {
 	ID            string     `gorm:"primaryKey"    json:"id"`
 	LastHeartbeat time.Time  `json:"lastHeartbeat"`
@@ -34,6 +41,7 @@ type Preparation struct {
 	DeleteAfterExport bool          `json:"deleteAfterExport"` // DeleteAfterExport is a flag that indicates whether the source files should be deleted after export.
 	MaxSize           int64         `json:"maxSize"`
 	PieceSize         int64         `json:"pieceSize"`
+	MinPieceSize      int64         `json:"minPieceSize"` // Minimum piece size for the preparation, applies only to DAG and remainder pieces
 	NoInline          bool          `json:"noInline"`
 	NoDag             bool          `json:"noDag"`
 
@@ -252,6 +260,7 @@ type CarID uint32
 type Car struct {
 	ID          CarID      `cbor:"-"                    gorm:"primaryKey"                                        json:"id"                                  table:"verbose"`
 	CreatedAt   time.Time  `cbor:"-"                    json:"createdAt"                                         table:"verbose;format:2006-01-02 15:04:05"`
+	PieceType   PieceType  `cbor:"0,keyasint,omitempty" json:"pieceType"                                         swaggertype:"string"` // PieceType indicates whether this is a data piece or DAG piece
 	PieceCID    CID        `cbor:"1,keyasint,omitempty" gorm:"column:piece_cid;index;type:bytes;size:255"        json:"pieceCid"                            swaggertype:"string"`
 	PieceSize   int64      `cbor:"2,keyasint,omitempty" json:"pieceSize"`
 	RootCID     CID        `cbor:"3,keyasint,omitempty" gorm:"column:root_cid;type:bytes"                        json:"rootCid"                             swaggertype:"string"`
@@ -319,3 +328,12 @@ func (c CarBlock) BlockLength() int32 {
 
 	return c.blockLength
 }
+
+// GetMinPieceSize returns the minimum piece size for the preparation, with a fallback to 1MiB if not set.
+// This ensures backward compatibility with older preparations that don't have minPieceSize set.
+func (p *Preparation) GetMinPieceSize() int64 {
+	if p.MinPieceSize == 0 {
+		return 1 << 20 // 1MiB
+	}
+	return p.MinPieceSize
+}