fix: skip rewriting CAR blocks that are already stored

Author: mennatnaga

internal/datastore/blockstore_test.go

@@ -15,6 +15,7 @@ import (
 	"testing"
 	"time"
 
+	blocks "github.com/ipfs/go-block-format"
 	"github.com/stretchr/testify/require"
 
 	"github.com/sourcenetwork/corekv/memory"
@@ -80,6 +81,48 @@ func TestIsMerged_DeletedBlockIsNotMerged(t *testing.T) {
 	require.False(t, merged)
 }
 
+// A CAR import re-puts blocks the node already holds. Storing one unconditionally stamps a fresh
+// to-merge marker over a block that already merged, which reads as an abandoned fetch and leaves
+// the orphan sweep free to delete a block a live document still links to.
+func TestIsMerged_ReputtingAMergedBlockLeavesItMerged(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		put  func(context.Context, Blockstore, blocks.Block) error
+	}{
+		{
+			name: "Put",
+			put:  func(ctx context.Context, bs Blockstore, b blocks.Block) error { return bs.Put(ctx, b) },
+		},
+		{
+			name: "PutMany",
+			put: func(ctx context.Context, bs Blockstore, b blocks.Block) error {
+				return bs.PutMany(ctx, []blocks.Block{b})
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			rootstore := memory.NewDatastore(ctx)
+
+			blockNS := namespace.Wrap(rootstore, []byte{blockStoreKey})
+			block := putMarkedBlock(t, ctx, blockNS, []byte("merged then re-imported"),
+				newToMergeValue(time.Unix(1_700_000_000, 0)))
+
+			blockstore := BlockstoreFrom(rootstore, immutable.None[int]())
+			require.NoError(t, blockstore.MarkAsMerged(ctx, block.Cid()))
+			merged, err := blockstore.IsMerged(ctx, block.Cid())
+			require.NoError(t, err)
+			require.True(t, merged)
+
+			require.NoError(t, tc.put(ctx, P2PBlockstoreFrom(rootstore, immutable.None[int]()), block))
+
+			merged, err = blockstore.IsMerged(ctx, block.Cid())
+			require.NoError(t, err)
+			require.True(t, merged, "storing a block the node already holds must not unmerge it")
+		})
+	}
+}
+
 func TestIsMerged_UnmergedAndAbsentBlocks(t *testing.T) {
 	ctx := context.Background()
 	rootstore := memory.NewDatastore(ctx)

internal/db/p2p/car.go

@@ -224,7 +224,10 @@ func (p *P2P) importCAR(ctx context.Context, carData []byte) (*coreblock.Block,
 		return nil, p.carImportFailure(reasonNoRoots, written, ErrEmptyCARRoots)
 	}
 
-	bstore := datastore.BlindWriteP2PBlockstoreFrom(p.db.Rootstore(), immutable.None[int]())
+	// Content-addressed field blocks recur across documents that share a field value, so a
+	// CAR routinely contains blocks already stored. The guarded store skips those instead of
+	// rewriting the block and re-stamping a to-merge marker on one that already merged.
+	bstore := datastore.P2PBlockstoreFrom(p.db.Rootstore(), immutable.None[int]())
 	encStore := datastore.EncstoreFrom(p.db.Rootstore())
 	var rootBlock *coreblock.Block