separate aggregate piece_ref, no pdp ref for aggregate

Author: LexLuthr

cmd/curio/storacha-migration-aggregate.go

@@ -34,6 +34,7 @@ import (
 )
 
 const storachaAggregateDefaultPieceSize = abi.PaddedPieceSize(1 << 30)
+const storachaMigrationAggregateDataURL = "storacha-migration-aggregate"
 
 // storachaAggregatePieceSize is a variable only so tests can exercise the
 // recovery paths without writing 1 GiB aggregate files. Production uses the
@@ -1771,7 +1772,7 @@ func recoverStorachaAggregateFromFinal(ctx context.Context, db *harmonydb.DB, si
 			return false, err
 		}
 
-		imported, err := recoverFinalStorachaPiece(ctx, db, si, storageID, targetPath, row)
+		imported, err := recoverFinalStorachaPiece(ctx, db, si, storageID, targetPath, row, false)
 		if err != nil {
 			return false, err
 		}
@@ -2045,7 +2046,7 @@ func importStagedStorachaAggregateFile(ctx context.Context, db *harmonydb.DB, si
 		return xerrors.Errorf("aggregate staging file %s size mismatch: expected %d, got %d; cleanup manually before retrying", stagingPath, verified.RawSize, info.Size())
 	}
 
-	result, err := importStagedStorachaPiece(ctx, db, si, storageID, targetPath, stagingPath)
+	result, err := importStagedStorachaPiece(ctx, db, si, storageID, targetPath, stagingPath, false)
 	if err != nil {
 		return err
 	}

cmd/curio/storacha-migration-aggregate_test.go

@@ -156,6 +156,7 @@ func TestStorachaAggregateRecoversFromFinalFileAndFixesDB(t *testing.T) {
 	tempPath, verified := createStorachaAggregateTempFileForTest(t, env, workDir, group)
 	pieceID := insertAggregateParkedPiece(t, env, verified, false)
 	assertStorachaRefs(t, env, pieceID, 0)
+	assertStorachaRefsWithDataURL(t, env, pieceID, storachaMigrationAggregateDataURL, 0)
 	assertSectorLocation(t, env, pieceID, 0)
 
 	finalPath := storachaFinalPiecePath(env.targetDir, pieceID)
@@ -166,8 +167,9 @@ func TestStorachaAggregateRecoversFromFinalFileAndFixesDB(t *testing.T) {
 
 	pp := parkedPieceByID(t, env, pieceID)
 	require.True(t, pp.complete)
-	assertStorachaRefs(t, env, pieceID, 1)
-	assertPDPRefs(t, env, pieceID, group.PieceCIDV1, 1)
+	assertStorachaRefs(t, env, pieceID, 0)
+	assertStorachaRefsWithDataURL(t, env, pieceID, storachaMigrationAggregateDataURL, 1)
+	assertPDPRefs(t, env, pieceID, group.PieceCIDV1, 0)
 	assertSectorLocation(t, env, pieceID, 1)
 
 	complete := assertStorachaAggregateImported(t, env, workDir, group)
@@ -407,8 +409,9 @@ func assertStorachaAggregateImported(t *testing.T, env storachaMigrationTestEnv,
 
 	pp := parkedPieceByID(t, env, rows[0].ID)
 	require.True(t, pp.complete)
-	assertStorachaRefs(t, env, rows[0].ID, 1)
-	assertPDPRefs(t, env, rows[0].ID, group.PieceCIDV1, 1)
+	assertStorachaRefs(t, env, rows[0].ID, 0)
+	assertStorachaRefsWithDataURL(t, env, rows[0].ID, storachaMigrationAggregateDataURL, 1)
+	assertPDPRefs(t, env, rows[0].ID, group.PieceCIDV1, 0)
 	assertSectorLocation(t, env, rows[0].ID, 1)
 
 	_, err = verifyStorachaAggregateFileForTest(storachaFinalPiecePath(env.targetDir, rows[0].ID), pcidV1, abi.PaddedPieceSize(group.PaddedSize))

cmd/curio/storacha-migration.go

@@ -217,7 +217,7 @@ func runImportPieces(ctx context.Context, db *harmonydb.DB, si paths.SectorIndex
 				continue
 			}
 
-			result, err := importStagedStorachaPiece(ctx, db, si, storageID, targetPath, filepath.Join(staging, entry.Name()))
+			result, err := importStagedStorachaPiece(ctx, db, si, storageID, targetPath, filepath.Join(staging, entry.Name()), true)
 			if err != nil {
 				return out, err
 			}
@@ -276,7 +276,7 @@ func runImportPieces(ctx context.Context, db *harmonydb.DB, si paths.SectorIndex
 			return out, xerrors.Errorf("renaming old path: %w", err)
 		}
 
-		result, err := importStagedStorachaPiece(ctx, db, si, storageID, targetPath, stagingPath)
+		result, err := importStagedStorachaPiece(ctx, db, si, storageID, targetPath, stagingPath, true)
 		if err != nil {
 			return out, err
 		}
@@ -334,13 +334,15 @@ func storachaPieceInfoFromFileName(name string) (*mk20.PieceInfo, bool, error) {
 
 // importStagedStorachaPiece finishes one target/storacha-staging/<cid>.car.
 // It first creates or reuses parked_pieces/refs in a committed DB transaction.
+// makePDPPieceRef keeps import-pieces behavior unchanged while letting aggregate
+// imports use the aggregate migration data_url and skip the public PDP piece ref.
 // If an existing incomplete parked_piece still has a live task, it returns
 // Imported=false and leaves the staged file untouched for a later retry.
 // Otherwise it consumes the staged file: remove it if the final
 // target/piece/s-t00-<id> already exists, or rename it to that final path. It
 // then declares the piece location and marks parked_pieces.complete unless the
 // row was already complete.
-func importStagedStorachaPiece(ctx context.Context, db *harmonydb.DB, si paths.SectorIndex, storageID storiface.ID, targetPath, stagingPath string) (storachaImportResult, error) {
+func importStagedStorachaPiece(ctx context.Context, db *harmonydb.DB, si paths.SectorIndex, storageID storiface.ID, targetPath, stagingPath string, makePDPPieceRef bool) (storachaImportResult, error) {
 	info, err := os.Stat(stagingPath)
 	if err != nil {
 		return storachaImportResult{}, xerrors.Errorf("checking staging file: %w", err)
@@ -383,7 +385,7 @@ func importStagedStorachaPiece(ctx context.Context, db *harmonydb.DB, si paths.S
 			return true, nil
 		}
 
-		if err := ensureStorachaRefsTx(tx, parkedPieceID, pi); err != nil {
+		if err := ensureStorachaRefsTx(tx, parkedPieceID, pi, makePDPPieceRef); err != nil {
 			return false, err
 		}
 
@@ -474,7 +476,7 @@ func recoverFinalStorachaPieces(ctx context.Context, db *harmonydb.DB, si paths.
 			break
 		}
 
-		imported, err := recoverFinalStorachaPiece(ctx, db, si, storageID, targetPath, row)
+		imported, err := recoverFinalStorachaPiece(ctx, db, si, storageID, targetPath, row, true)
 		if err != nil {
 			return err
 		}
@@ -510,7 +512,7 @@ func storachaPieceCIDV2FromRow(row storachaFinalRecoveryRow) (string, error) {
 // DB row, reuses the same claim/ref checks as staged import, then declares the
 // final file and marks the parked piece complete. It returns false when the
 // final file is missing or a live task still owns the parked_piece.
-func recoverFinalStorachaPiece(ctx context.Context, db *harmonydb.DB, si paths.SectorIndex, storageID storiface.ID, targetPath string, row storachaFinalRecoveryRow) (bool, error) {
+func recoverFinalStorachaPiece(ctx context.Context, db *harmonydb.DB, si paths.SectorIndex, storageID storiface.ID, targetPath string, row storachaFinalRecoveryRow, makePDPPieceRef bool) (bool, error) {
 	finalPath := storachaFinalPiecePath(targetPath, row.ID)
 	info, err := os.Stat(finalPath)
 	if os.IsNotExist(err) {
@@ -542,7 +544,7 @@ func recoverFinalStorachaPiece(ctx context.Context, db *harmonydb.DB, si paths.S
 		if !state.UseStagedFile {
 			return true, nil
 		}
-		if err := ensureStorachaRefsTx(tx, row.ID, pi); err != nil {
+		if err := ensureStorachaRefsTx(tx, row.ID, pi, makePDPPieceRef); err != nil {
 			return false, err
 		}
 		return true, nil
@@ -617,12 +619,15 @@ func claimExistingStorachaPieceTx(tx *harmonydb.Tx, parkedPieceID int64) (storac
 	return state, nil
 }
 
-// ensureStorachaRefsTx enforces the intended one-to-one shape for this import.
-// It locks storacha-migration parked_piece_refs for this parked piece, fails if
-// more than one exists, inserts the missing ref if needed, then does the same
-// for pdp_piecerefs through the unique piece_ref relation. Existing PDP refs
-// must already point at the same service and piece CID.
-func ensureStorachaRefsTx(tx *harmonydb.Tx, parkedPieceID int64, pi *mk20.PieceInfo) error {
+// ensureStorachaRefsTx enforces the intended one-to-one shape for this migration
+// mode. Direct imports create storacha-migration refs and PDP refs; aggregate
+// imports create storacha-migration-aggregate refs and intentionally skip PDP refs.
+func ensureStorachaRefsTx(tx *harmonydb.Tx, parkedPieceID int64, pi *mk20.PieceInfo, makePDPPieceRef bool) error {
+	dataURL := storachaMigrationAggregateDataURL
+	if makePDPPieceRef {
+		dataURL = storachaMigrationDataURL
+	}
+
 	var refs []struct {
 		RefID int64 `db:"ref_id"`
 	}
@@ -633,7 +638,7 @@ func ensureStorachaRefsTx(tx *harmonydb.Tx, parkedPieceID int64, pi *mk20.PieceI
 		  AND data_url = $2
 		  AND long_term = TRUE
 		ORDER BY ref_id
-		FOR UPDATE`, parkedPieceID, storachaMigrationDataURL)
+		FOR UPDATE`, parkedPieceID, dataURL)
 	if err != nil {
 		return xerrors.Errorf("query storacha parked piece refs: %w", err)
 	}
@@ -648,18 +653,26 @@ func ensureStorachaRefsTx(tx *harmonydb.Tx, parkedPieceID int64, pi *mk20.PieceI
 		err = tx.QueryRow(`
 			INSERT INTO parked_piece_refs (piece_id, data_url, long_term)
 			VALUES ($1, $2, TRUE)
-			RETURNING ref_id`, parkedPieceID, storachaMigrationDataURL).Scan(&pieceRef)
+			RETURNING ref_id`, parkedPieceID, dataURL).Scan(&pieceRef)
 		if err != nil {
 			return xerrors.Errorf("insert storacha parked piece ref: %w", err)
 		}
 	}
 
+	if !makePDPPieceRef {
+		return nil
+	}
+
+	return ensurePDPPieceRef(tx, pieceRef, pi)
+}
+
+func ensurePDPPieceRef(tx *harmonydb.Tx, pieceRef int64, pi *mk20.PieceInfo) error {
 	var pdpRefs []struct {
 		ID       int64  `db:"id"`
 		Service  string `db:"service"`
 		PieceCID string `db:"piece_cid"`
 	}
-	err = tx.Select(&pdpRefs, `
+	err := tx.Select(&pdpRefs, `
 		SELECT id, service, piece_cid
 		FROM pdp_piecerefs
 		WHERE piece_ref = $1

cmd/curio/storacha-migration_test.go

@@ -666,6 +666,11 @@ func parkedPieceByID(t *testing.T, env storachaMigrationTestEnv, pieceID int64)
 
 func storachaRefIDs(t *testing.T, env storachaMigrationTestEnv, pieceID int64) []int64 {
 	t.Helper()
+	return storachaRefIDsForDataURL(t, env, pieceID, storachaMigrationDataURL)
+}
+
+func storachaRefIDsForDataURL(t *testing.T, env storachaMigrationTestEnv, pieceID int64, dataURL string) []int64 {
+	t.Helper()
 
 	var rows []struct {
 		RefID int64 `db:"ref_id"`
@@ -677,7 +682,7 @@ func storachaRefIDs(t *testing.T, env storachaMigrationTestEnv, pieceID int64) [
 		  AND data_url = $2
 		  AND long_term = TRUE
 		ORDER BY ref_id
-	`, pieceID, storachaMigrationDataURL)
+	`, pieceID, dataURL)
 	require.NoError(t, err)
 
 	out := make([]int64, 0, len(rows))
@@ -707,6 +712,11 @@ func assertStorachaRefs(t *testing.T, env storachaMigrationTestEnv, pieceID int6
 	require.Len(t, storachaRefIDs(t, env, pieceID), expected)
 }
 
+func assertStorachaRefsWithDataURL(t *testing.T, env storachaMigrationTestEnv, pieceID int64, dataURL string, expected int) {
+	t.Helper()
+	require.Len(t, storachaRefIDsForDataURL(t, env, pieceID, dataURL), expected)
+}
+
 func assertPDPRefs(t *testing.T, env storachaMigrationTestEnv, pieceID int64, pieceCID string, expected int) {
 	t.Helper()
 	refs := pdpRefsForPiece(t, env, pieceID)