retrieve nodes from remote peers

experiment/serf_crdt.go

@@ -125,10 +125,16 @@ func main() {
 	logger := slog.New(tint.NewHandler(os.Stdout, &tint.Options{
 		AddSource: true,
 		Level:     slog.LevelDebug,
+		//Level:      slog.LevelInfo,
 		TimeFormat: time.RFC3339Nano,
 	}))
 	slog.SetDefault(logger)
 
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	// A channel to signal that shutdown is done.
+	done := make(chan bool, 1)
+
 	config := createSerfAgentConfig(*name, *bindAddr, *rpcAddr)
 	serfAgent, err := createSerfAgent(config)
 	if err != nil {
@@ -140,57 +146,72 @@ func main() {
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
-	// Ideally, the broadcaster should be registered as an event handler before starting the agent.
-	// However, we need the agent.serf to be initialized which is done in agent.Start().
-	broadcaster := NewSerfBroadcaster(ctx, serfAgent.Serf())
-	serfAgent.RegisterEventHandler(broadcaster)
-
-	sigs := make(chan os.Signal, 1)
-	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
-	// A channel to signal that shutdown is done.
-	done := make(chan bool, 1)
-
 	localStore, err := badger.NewDatastore(*storeDir, nil) // default options
 	if err != nil {
 		panic(err)
 	}
-	syncer := newDAGSyncer(localStore, ds.NewKey("/node"))
+	syncer := newDAGSyncer(localStore, ds.NewKey("/node"), serfAgent.Serf())
+
+	// Ideally, the broadcaster should be registered as an event handler before starting the agent.
+	// However, we need the agent.serf to be initialized which is done in agent.Start().
+	broadcaster := NewSerfBroadcaster(ctx, serfAgent.Serf())
+	serfAgent.RegisterEventHandler(broadcaster)
+	serfAgent.RegisterEventHandler(syncer)
 
 	opts := crdt.DefaultOptions()
 	opts.Logger = newIPFSLogger(logger)
+	//opts.MultiHeadProcessing = true
+	// TODO: debug why the heads count may grow on the receiving side if the backlog is huge and perhaps when the node
+	//  is shutdowned before processing all the backlog.
 	store, err := crdt.New(localStore, ds.NewKey("/"), syncer, broadcaster, opts)
 	if err != nil {
 		panic(err)
 	}
 
-	if *runTick {
+	//ticker := time.NewTicker(10 * time.Millisecond)
-		ticker := time.NewTicker(10 * time.Second)
+	ticker := time.NewTicker(3 * time.Second)
 	go func() {
 		for {
 			select {
 			case t := <-ticker.C:
+				if *runTick {
 					err = store.Put(ctx, ds.NewKey("/tick"), []byte(t.String()))
 					if err != nil {
 						slog.Error("Put /tick", "error", err)
 					}
+					value, err := store.Get(ctx, ds.NewKey("/tick"))
+					if err != nil {
+						slog.Error("Get /tick", "error", err)
+					}
+					slog.Info("Get /tick", "value", string(value))
+					stats := store.InternalStats()
+					slog.Info("CRDT store stats", "stats", stats, "heads_count", len(stats.Heads))
+				} else {
+					value, err := store.Get(ctx, ds.NewKey("/tick"))
+					if err != nil {
+						slog.Error("Get /tick", "error", err)
+					}
+					slog.Info("Get /tick", "value", string(value))
+					stats := store.InternalStats()
+					slog.Info("CRDT store stats", "stats", stats, "heads_count", len(stats.Heads))
+				}
 			case <-ctx.Done():
 				return
 			}
 		}
 	}()
-	}
 
 	//err = store.Put(ctx, ds.NewKey("/test3"), []byte("hello3"))
 	//if err != nil {
 	//	panic(err)
 	//}
-	v, err := store.Get(ctx, ds.NewKey("/test"))
+	//v, err := store.Get(ctx, ds.NewKey("/test"))
-	if err != nil {
+	//if err != nil {
-		slog.Error("Get /test", "error", err)
+	//	slog.Error("Get /test", "error", err)
-	}
+	//}
-	slog.Info("Get /test", "value", string(v))
+	//slog.Info("Get /test", "value", string(v))
 
-	_ = store.PrintDAG()
+	//_ = store.PrintDAG()
 
 	// Start a goroutine to handle signals.
 	go func() {

experiment/syncer.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/hashicorp/serf/serf"
 	"github.com/ipfs/boxo/datastore/dshelp"
 	dag "github.com/ipfs/boxo/ipld/merkledag"
 	"github.com/ipfs/go-cid"
@@ -19,41 +20,110 @@ type dagSyncer struct {
 	// Persistent storage for the nodes.
 	store     ds.Datastore
 	namespace ds.Key
+	serf      *serf.Serf
 }
 
-func newDAGSyncer(store ds.Datastore, namespace ds.Key) *dagSyncer {
+func newDAGSyncer(store ds.Datastore, namespace ds.Key, serf *serf.Serf) *dagSyncer {
 	return &dagSyncer{
 		store:     store,
 		namespace: namespace,
+		serf:      serf,
 	}
 }
 
 func (d *dagSyncer) Get(ctx context.Context, cid cid.Cid) (ipld.Node, error) {
 	slog.Debug("Getting node", "cid", cid)
-	nodeBytes, err := d.store.Get(ctx, d.nodeKey(cid))
+	node, err := d.getNode(ctx, cid)
+	if err == nil {
+		slog.Debug("Found node in local store", "cid", cid)
+		return node, nil
+	}
+	if !errors.Is(err, ds.ErrNotFound) {
+		return nil, err
+	}
+
+	// The node it not found in the local store. Try to retrieve it from remote peers.
+	// TODO: exclude the local peer from the query.
+	query, err := d.serf.Query("get-node", cid.Bytes(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("query node from peers: %w", err)
+	}
+	slog.Debug("Queried node from peers", "cid", cid, "deadline", query.Deadline())
+
+	responded := false
+	for !responded {
+		select {
+		case resp, ok := <-query.ResponseCh():
+			if !ok {
+				// The query has finished and no response was received.
+				slog.Warn("Query for node finished without response", "cid", cid)
+				return nil, ipld.ErrNotFound{Cid: cid}
+			}
+			if resp.From == d.serf.LocalMember().Name {
+				continue
+			}
+			slog.Debug("Received node from peer", "cid", cid, "peer", resp.From)
+			responded = true
+			query.Close()
+
+			node, err = nodeFromBytes(resp.Payload)
+			if err != nil {
+				return nil, err
+			}
+			// Ensure the received node is actually the requested node.
+			if node.Cid() != cid {
+				return nil, fmt.Errorf("received node CID %s does not match requested CID %s", node.Cid(), cid)
+			}
+			if err = d.persistNode(ctx, node); err != nil {
+				return nil, err
+			}
+			return node, nil
+		case <-ctx.Done():
+			query.Close()
+			return nil, ctx.Err()
+		}
+	}
+	// This return should never be reached.
+	return nil, ipld.ErrNotFound{Cid: cid}
+}
+
+func (d *dagSyncer) getNode(ctx context.Context, cid cid.Cid) (*dag.ProtoNode, error) {
+	bytes, err := d.store.Get(ctx, d.nodeKey(cid))
 	if err != nil {
 		if errors.Is(err, ds.ErrNotFound) {
-			// TODO: try to retrieve the node from the peers.
+			return nil, ds.ErrNotFound
-			return nil, ipld.ErrNotFound{Cid: cid}
 		}
 		return nil, fmt.Errorf("get node %s from local store: %w", cid, err)
 	}
-	protoNode, err := dag.DecodeProtobuf(nodeBytes)
+	return nodeFromBytes(bytes)
-	if err != nil {
-		return nil, fmt.Errorf("decode node from protobuf: %w", err)
 }
-	// CID is lazily computed from the node content. Ensure the node uses CIDv1.
-	if err = protoNode.SetCidBuilder(dag.V1CidPrefix()); err != nil {
-		return nil, fmt.Errorf("set CIDv1 on node: %w", err)
-	}
-	slog.Debug("Retrieved node from local store", "cid", protoNode.Cid())
 
-	return protoNode, nil
+func (d *dagSyncer) persistNode(ctx context.Context, node *dag.ProtoNode) error {
+	bytes, err := node.EncodeProtobuf(false)
+	if err != nil {
+		return fmt.Errorf("encode node to protobuf: %w", err)
+	}
+	if err = d.store.Put(ctx, d.nodeKey(node.Cid()), bytes); err != nil {
+		return fmt.Errorf("put node %s in local store: %w", node.String(), err)
+	}
+	slog.Debug("Persisted node in local store", "cid", node.Cid(), "size", len(bytes))
+	return nil
 }
 
 func (d *dagSyncer) GetMany(ctx context.Context, cids []cid.Cid) <-chan *ipld.NodeOption {
-	//TODO implement me
+	ch := make(chan *ipld.NodeOption)
-	panic("implement me")
+	go func() {
+		defer close(ch)
+		for _, cid := range cids {
+			node, err := d.Get(ctx, cid)
+			if err != nil {
+				ch <- &ipld.NodeOption{Err: err}
+			} else {
+				ch <- &ipld.NodeOption{Node: node}
+			}
+		}
+	}()
+	return ch
 }
 
 func (d *dagSyncer) Add(ctx context.Context, node ipld.Node) error {
@@ -62,19 +132,11 @@ func (d *dagSyncer) Add(ctx context.Context, node ipld.Node) error {
 	if !ok {
 		return fmt.Errorf("node is not a ProtoNode")
 	}
-	nodeBytes, err := protoNode.EncodeProtobuf(false)
+	if err := d.persistNode(ctx, protoNode); err != nil {
-	if err != nil {
+		return err
-		return fmt.Errorf("encode node to protobuf: %w", err)
 	}
-
-	if err = d.store.Put(ctx, d.nodeKey(node.Cid()), nodeBytes); err != nil {
-		return fmt.Errorf("put node %s in local store: %w", node, err)
-	}
-	slog.Debug("Persisted node in local store", "cid", node.Cid(), "size", len(nodeBytes))
-
 	// TODO: Think about broadcasting the new node to the peers to not require each peer to query the node
 	//  when a new CID is broadcasted.
-
 	return nil
 }
 
@@ -90,8 +152,56 @@ func (d *dagSyncer) RemoveMany(ctx context.Context, cids []cid.Cid) error {
 	panic("implement me")
 }
 
+// TODO: perhaps this should not be part of the syncer/DAG service?
+func (d *dagSyncer) HandleEvent(event serf.Event) {
+	query, ok := event.(*serf.Query)
+	if !ok || query.Name != "get-node" {
+		// Ignore non-get-node queries.
+		return
+	}
+	_, cid, err := cid.CidFromBytes(query.Payload)
+	if err != nil {
+		slog.Error("Decode CID from query payload", "error", err)
+		return
+	}
+	slog.Debug("Received get-node query", "cid", cid)
+	log := slog.With("cid", cid)
+
+	node, err := d.getNode(context.Background(), cid)
+	if err != nil {
+		if !errors.Is(err, ds.ErrNotFound) {
+			log.Error("Get node from local store", "error", err)
+			return
+		}
+		// The node is not found in the local store. Ignore the query.
+		log.Debug("Node not found in local store")
+		return
+	}
+	bytes, err := node.EncodeProtobuf(false)
+	if err != nil {
+		log.Error("Encode node to protobuf", "error", err)
+	}
+	if err = query.Respond(bytes); err != nil {
+		log.Error("Respond to get-node query", "error", err)
+		return
+	}
+	log.Debug("Responded to get-node query")
+}
+
 func (d *dagSyncer) nodeKey(cid cid.Cid) ds.Key {
 	return d.namespace.Child(dshelp.MultihashToDsKey(cid.Hash()))
 }
 
 var _ ipld.DAGService = (*dagSyncer)(nil)
+
+func nodeFromBytes(nodeBytes []byte) (*dag.ProtoNode, error) {
+	node, err := dag.DecodeProtobuf(nodeBytes)
+	if err != nil {
+		return nil, fmt.Errorf("decode node from protobuf: %w", err)
+	}
+	// CID is lazily computed from the node content. Ensure the node uses CIDv1.
+	if err = node.SetCidBuilder(dag.V1CidPrefix()); err != nil {
+		return nil, fmt.Errorf("set CIDv1 on node: %w", err)
+	}
+	return node, nil
+}