diff --git a/experiment/serf_crdt.go b/experiment/serf_crdt.go index d9eb5621..ec9cec13 100644 --- a/experiment/serf_crdt.go +++ b/experiment/serf_crdt.go @@ -123,12 +123,18 @@ func main() { flag.Parse() logger := slog.New(tint.NewHandler(os.Stdout, &tint.Options{ - AddSource: true, - Level: slog.LevelDebug, + 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.Second) - go func() { - for { - select { - case t := <-ticker.C: + //ticker := time.NewTicker(10 * time.Millisecond) + 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) } - case <-ctx.Done(): - return + 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")) - if err != nil { - slog.Error("Get /test", "error", err) - } - slog.Info("Get /test", "value", string(v)) + //v, err := store.Get(ctx, ds.NewKey("/test")) + //if err != nil { + // slog.Error("Get /test", "error", err) + //} + //slog.Info("Get /test", "value", string(v)) - _ = store.PrintDAG() + //_ = store.PrintDAG() // Start a goroutine to handle signals. go func() { diff --git a/experiment/syncer.go b/experiment/syncer.go index f685f510..244d99cb 100644 --- a/experiment/syncer.go +++ b/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, ipld.ErrNotFound{Cid: cid} + return nil, ds.ErrNotFound } return nil, fmt.Errorf("get node %s from local store: %w", cid, err) } - protoNode, 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 = 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 nodeFromBytes(bytes) +} - 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 - panic("implement me") + ch := make(chan *ipld.NodeOption) + 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 != nil { - return fmt.Errorf("encode node to protobuf: %w", err) + if err := d.persistNode(ctx, protoNode); err != nil { + return 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 +}