test walk sync process

go.mod

@@ -185,10 +185,12 @@ require (
 	github.com/ipfs/go-ipfs-delay v0.0.1 // indirect
 	github.com/ipfs/go-ipfs-pq v0.0.3 // indirect
 	github.com/ipfs/go-ipfs-util v0.0.3 // indirect
+	github.com/ipfs/go-ipld-legacy v0.2.1 // indirect
 	github.com/ipfs/go-log v1.0.5 // indirect
 	github.com/ipfs/go-metrics-interface v0.0.1 // indirect
 	github.com/ipfs/go-peertaskqueue v0.8.1 // indirect
 	github.com/ipfs/kubo v0.25.0 // indirect
+	github.com/ipld/go-codec-dagpb v1.6.0 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/jmhodges/levigo v1.0.0 // indirect

go.sum

@@ -1166,6 +1166,8 @@ github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49u
 github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
 github.com/vito/go-sse v1.0.0 h1:e6/iTrrvy8BRrOwJwmQmlndlil+TLdxXvHi55ZDzH6M=
 github.com/vito/go-sse v1.0.0/go.mod h1:2wkcaQ+jtlZ94Uve8gYZjFpL68luAjssTINA2hpgcZs=
+github.com/warpfork/go-testmark v0.12.1 h1:rMgCpJfwy1sJ50x0M0NgyphxYYPMOODIJHhsXyEHU0s=
+github.com/warpfork/go-testmark v0.12.1/go.mod h1:kHwy7wfvGSPh1rQJYKayD4AbtNaeyZdcGi9tNJTaa5Y=
 github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0 h1:GDDkbFiaK8jsSDJfjId/PEGEShv6ugrt4kYsC5UIDaQ=
 github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=
 github.com/wasmerio/wasmer-go v1.0.4 h1:MnqHoOGfiQ8MMq2RF6wyCeebKOe84G88h5yv+vmxJgs=

net/server.go

@@ -17,7 +17,8 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/ipfs/go-cid"
+	"github.com/ipfs/boxo/ipld/merkledag"
+	cid "github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p/core/event"
 	libpeer "github.com/libp2p/go-libp2p/core/peer"
 	"github.com/sourcenetwork/corelog"
@@ -28,9 +29,9 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/datastore/badger/v4"
 	"github.com/sourcenetwork/defradb/errors"
 	"github.com/sourcenetwork/defradb/events"
-	coreblock "github.com/sourcenetwork/defradb/internal/core/block"
 	pb "github.com/sourcenetwork/defradb/net/pb"
 )
 
@@ -150,7 +151,7 @@ func (s *server) PushLog(ctx context.Context, req *pb.PushLogRequest) (*pb.PushL
 	if err != nil {
 		return nil, err
 	}
-	cid, err := cid.Cast(req.Body.Cid)
+	headCID, err := cid.Cast(req.Body.Cid)
 	if err != nil {
 		return nil, err
 	}
@@ -177,38 +178,29 @@ func (s *server) PushLog(ctx context.Context, req *pb.PushLogRequest) (*pb.PushL
 		}
 	}()
 
-	// check if we already have this block
-	exists, err := s.peer.db.Blockstore().Has(ctx, cid)
-	if err != nil {
-		return nil, NewErrCheckingForExistingBlock(err, cid.String())
-	}
-	if exists {
-		return &pb.PushLogReply{}, nil
+	onError := merkledag.OnError(func(c cid.Cid, err error) error {
+		if errors.Is(err, badger.ErrTxnConflict) {
+			return nil // transaction conflicts are fine to ignore
+		}
+		return err
+	})
+	visit := func(c cid.Cid) bool {
+		has, _ := s.peer.db.Blockstore().Has(ctx, c)
+		return !has
 	}
-
-	block, err := coreblock.GetFromBytes(req.Body.Log.Block)
+	dagServ := merkledag.NewDAGService(s.peer.bserv)
+	walkOpts := []merkledag.WalkOption{onError, merkledag.Concurrent()}
+	err = merkledag.Walk(ctx, dagServ.GetLinks, headCID, visit, walkOpts...)
 	if err != nil {
 		return nil, err
 	}
 
-	bp := newBlockProcessor(ctx, s.peer)
-	err = bp.processRemoteBlock(ctx, block)
-	if err != nil {
-		log.ErrorContextE(
-			ctx,
-			"Failed to process remote block",
-			err,
-			corelog.String("DocID", docID.String()),
-			corelog.Any("CID", cid),
-		)
-	}
-	bp.wg.Wait()
 	if s.peer.db.Events().DAGMerges.HasValue() {
 		wg := &sync.WaitGroup{}
 		wg.Add(1)
 		s.peer.db.Events().DAGMerges.Value().Publish(events.DAGMerge{
 			DocID:      docID.String(),
-			Cid:        cid,
+			Cid:        headCID,
 			SchemaRoot: string(req.Body.SchemaRoot),
 			Wg:         wg,
 		})