experiment: Serf transport for go-ds-crdt

2026-08-26 11:03:34 +00:00 · 2024-08-07 13:51:53 +10:00
parent b4606b939b
commit e6d8a33928
5 changed files with 509 additions and 0 deletions
@@ -0,0 +1,73 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/hashicorp/serf/serf"
+	crdt "github.com/ipfs/go-ds-crdt"
+	"log/slog"
+	"time"
+)
+
+// Implements the Broadcaster interface.
+type SerfBroadcaster struct {
+	ctx    context.Context
+	serf   *serf.Serf
+	nextCh chan []byte
+}
+
+// The broadcaster can be shut down by cancelling the given context. This must be done before Closing
+// the crdt.Datastore, otherwise things may hang.
+func NewSerfBroadcaster(ctx context.Context, serf *serf.Serf) *SerfBroadcaster {
+	return &SerfBroadcaster{
+		ctx:    ctx,
+		serf:   serf,
+		nextCh: make(chan []byte),
+	}
+}
+
+func (b *SerfBroadcaster) Broadcast(bytes []byte) error {
+	slog.Debug("Broadcasting head nodes to peers", "size", len(bytes))
+	// Other peers are not allowed to coalesce this event by name as the payload may differ.
+	// TODO: decode CRDTBroadcast from bytes and embed the nodes content in the payload as well to save round trip.
+	if err := b.serf.UserEvent("heads", bytes, false); err != nil {
+		return fmt.Errorf("broadcast heads event: %w", err)
+	}
+	return nil
+}
+
+func (b *SerfBroadcaster) Next() ([]byte, error) {
+	select {
+	case bytes := <-b.nextCh: // Blocks until a new heads event is received.
+		return bytes, nil
+	case <-b.ctx.Done():
+		return nil, crdt.ErrNoMoreBroadcast
+	}
+}
+
+func (b *SerfBroadcaster) HandleEvent(event serf.Event) {
+	select {
+	case <-b.ctx.Done():
+		return
+	default:
+	}
+
+	slog.Debug("Received event in broadcaster", "event", event)
+	userEvent, ok := event.(serf.UserEvent)
+	if !ok {
+		// Ignore non-user events.
+		return
+	}
+	if userEvent.Name != "heads" {
+		// Ignore non-heads user events.
+		return
+	}
+	start := time.Now()
+	select {
+	case b.nextCh <- userEvent.Payload:
+		slog.Debug("Handled heads event", "duration", time.Since(start))
+	case <-b.ctx.Done():
+	}
+}
+
+var _ crdt.Broadcaster = (*SerfBroadcaster)(nil)
@@ -0,0 +1,45 @@
+package main
+
+import (
+	"fmt"
+	"github.com/dgraph-io/badger/v3"
+	"log"
+	"time"
+)
+
+func customTimeEncoder(t time.Time) string {
+	return t.Format("2006-01-02T15:04:05.000000Z07:00")
+}
+
+func main() {
+	opts := badger.DefaultOptions("./store")
+	opts.Logger = nil // Disable the noisy Badger's default logger
+
+	db, err := badger.Open(opts)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer db.Close()
+
+	err = db.View(func(txn *badger.Txn) error {
+		opts := badger.DefaultIteratorOptions
+		opts.PrefetchSize = 10
+		it := txn.NewIterator(opts)
+		defer it.Close()
+		for it.Rewind(); it.Valid(); it.Next() {
+			item := it.Item()
+			k := item.Key()
+			err := item.Value(func(v []byte) error {
+				fmt.Printf("key=%s, value=%s, raw value=%v\n", k, v, v)
+				return nil
+			})
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+}
@@ -0,0 +1,73 @@
+package main
+
+import (
+	"fmt"
+	"github.com/ipfs/go-log/v2"
+	"log/slog"
+	"os"
+)
+
+// ipfsLogger is an slog logger that implements the IPFS go-log StandardLogger interface.
+type ipfsLogger struct {
+	log slog.Logger
+}
+
+func newIPFSLogger(l *slog.Logger) *ipfsLogger {
+	return &ipfsLogger{log: *l}
+}
+
+func (l *ipfsLogger) Debug(args ...interface{}) {
+	l.log.Debug(fmt.Sprint(args...))
+}
+
+func (l *ipfsLogger) Debugf(format string, args ...interface{}) {
+	l.log.Debug(fmt.Sprintf(format, args...))
+}
+
+func (l *ipfsLogger) Error(args ...interface{}) {
+	l.log.Error(fmt.Sprint(args...))
+}
+
+func (l *ipfsLogger) Errorf(format string, args ...interface{}) {
+	l.log.Error(fmt.Sprintf(format, args...))
+}
+
+func (l *ipfsLogger) Fatal(args ...interface{}) {
+	l.log.Error(fmt.Sprint(args...))
+	os.Exit(1)
+}
+
+func (l *ipfsLogger) Fatalf(format string, args ...interface{}) {
+	l.log.Error(fmt.Sprintf(format, args...))
+	os.Exit(1)
+}
+
+func (l *ipfsLogger) Info(args ...interface{}) {
+	l.log.Info(fmt.Sprint(args...))
+}
+
+func (l *ipfsLogger) Infof(format string, args ...interface{}) {
+	l.log.Info(fmt.Sprintf(format, args...))
+}
+
+func (l *ipfsLogger) Panic(args ...interface{}) {
+	msg := fmt.Sprint(args...)
+	l.log.Error(msg)
+	panic(msg)
+}
+
+func (l *ipfsLogger) Panicf(format string, args ...interface{}) {
+	msg := fmt.Sprintf(format, args...)
+	l.log.Error(msg)
+	panic(msg)
+}
+
+func (l *ipfsLogger) Warn(args ...interface{}) {
+	l.log.Warn(fmt.Sprint(args...))
+}
+
+func (l *ipfsLogger) Warnf(format string, args ...interface{}) {
+	l.log.Warn(fmt.Sprintf(format, args...))
+}
+
+var _ log.StandardLogger = (*ipfsLogger)(nil)
@@ -0,0 +1,221 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"github.com/hashicorp/memberlist"
+	"github.com/hashicorp/serf/cmd/serf/command/agent"
+	"github.com/hashicorp/serf/serf"
+	ds "github.com/ipfs/go-datastore"
+	badger "github.com/ipfs/go-ds-badger3"
+	crdt "github.com/ipfs/go-ds-crdt"
+	"github.com/lmittmann/tint"
+	"log/slog"
+	"net"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+func createSerfAgentConfig(name, bindAddr, rpcAddr string) *agent.Config {
+	config := agent.DefaultConfig()
+	config.NodeName = name
+	config.BindAddr = bindAddr
+	config.RPCAddr = rpcAddr
+
+	return config
+}
+
+func createSerfAgent(config *agent.Config) (*agent.Agent, error) {
+	bindIP, bindPort, err := config.AddrParts(config.BindAddr)
+	if err != nil {
+		return nil, fmt.Errorf("invalid bind address: %w", err)
+	}
+	serfConfig := serf.DefaultConfig()
+	switch config.Profile {
+	case "lan":
+		serfConfig.MemberlistConfig = memberlist.DefaultLANConfig()
+	case "wan":
+		serfConfig.MemberlistConfig = memberlist.DefaultWANConfig()
+	case "local":
+		serfConfig.MemberlistConfig = memberlist.DefaultLocalConfig()
+	default:
+		return nil, fmt.Errorf("unknown profile: %s", config.Profile)
+	}
+
+	serfConfig.MemberlistConfig.BindAddr = bindIP
+	serfConfig.MemberlistConfig.BindPort = bindPort
+	//serfConfig.MemberlistConfig.AdvertiseAddr = advertiseIP
+	//serfConfig.MemberlistConfig.AdvertisePort = advertisePort
+	//serfConfig.MemberlistConfig.SecretKey = encryptKey
+	serfConfig.NodeName = config.NodeName
+	serfConfig.Tags = config.Tags
+	serfConfig.SnapshotPath = config.SnapshotPath
+	serfConfig.ProtocolVersion = uint8(config.Protocol)
+	serfConfig.CoalescePeriod = 3 * time.Second
+	serfConfig.QuiescentPeriod = time.Second
+	serfConfig.QueryResponseSizeLimit = config.QueryResponseSizeLimit
+	serfConfig.QuerySizeLimit = config.QuerySizeLimit
+	serfConfig.UserEventSizeLimit = config.UserEventSizeLimit
+	serfConfig.UserCoalescePeriod = 3 * time.Second
+	serfConfig.UserQuiescentPeriod = time.Second
+	if config.ReconnectInterval != 0 {
+		serfConfig.ReconnectInterval = config.ReconnectInterval
+	}
+	if config.ReconnectTimeout != 0 {
+		serfConfig.ReconnectTimeout = config.ReconnectTimeout
+	}
+	if config.TombstoneTimeout != 0 {
+		serfConfig.TombstoneTimeout = config.TombstoneTimeout
+	}
+	serfConfig.EnableNameConflictResolution = !config.DisableNameResolution
+	if config.KeyringFile != "" {
+		serfConfig.KeyringFile = config.KeyringFile
+	}
+	serfConfig.RejoinAfterLeave = config.RejoinAfterLeave
+	if config.BroadcastTimeout != 0 {
+		serfConfig.BroadcastTimeout = config.BroadcastTimeout
+	}
+
+	serfAgent, err := agent.Create(config, serfConfig, os.Stdout)
+	if err != nil {
+		return nil, fmt.Errorf("setup the Serf agent: %w", err)
+	}
+	return serfAgent, nil
+}
+
+func startSerfAgent(config *agent.Config, serfAgent *agent.Agent) (*agent.AgentIPC, error) {
+	if err := serfAgent.Start(); err != nil {
+		return nil, fmt.Errorf("start Serf agent: %w", err)
+	}
+
+	// Setup the RPC listener
+	rpcListener, err := net.Listen("tcp", config.RPCAddr)
+	if err != nil {
+		return nil, fmt.Errorf("start RPC listener: %w", err)
+	}
+
+	// Start the IPC layer
+	logWriter := agent.NewLogWriter(512)
+	ipc := agent.NewAgentIPC(serfAgent, config.RPCAuthKey, rpcListener, os.Stdout, logWriter)
+
+	slog.Info("Started Serf agent",
+		"name", config.NodeName,
+		"bind_addr", config.BindAddr,
+		"rpc_addr", config.RPCAddr,
+		"encrypted", serfAgent.Serf().EncryptionEnabled(),
+		"snapshot", config.SnapshotPath != "",
+		"profile", config.Profile,
+		"compression", config.EnableCompression,
+	)
+
+	return ipc, nil
+}
+
+func main() {
+	name := flag.String("name", "", "Machine name")
+	bindAddr := flag.String("bind", "0.0.0.0:7946", "Bind address")
+	rpcAddr := flag.String("rpc-addr", "127.0.0.1:7373", "RPC address")
+	storeDir := flag.String("store-dir", "./store", "Store directory path")
+	runTick := flag.Bool("tick", false, "Periodically update /tick value")
+	flag.Parse()
+
+	logger := slog.New(tint.NewHandler(os.Stdout, &tint.Options{
+		AddSource:  true,
+		Level:      slog.LevelDebug,
+		TimeFormat: time.RFC3339Nano,
+	}))
+	slog.SetDefault(logger)
+
+	config := createSerfAgentConfig(*name, *bindAddr, *rpcAddr)
+	serfAgent, err := createSerfAgent(config)
+	if err != nil {
+		panic(err)
+	}
+	ipc, err := startSerfAgent(config, serfAgent)
+	if err != nil {
+		panic(err)
+	}
+
+	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"))
+
+	opts := crdt.DefaultOptions()
+	opts.Logger = newIPFSLogger(logger)
+	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:
+					err = store.Put(ctx, ds.NewKey("/tick"), []byte(t.String()))
+					if err != nil {
+						slog.Error("Put /tick", "error", err)
+					}
+				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))
+
+	_ = store.PrintDAG()
+
+	// Start a goroutine to handle signals.
+	go func() {
+		sig := <-sigs
+		slog.Info("Received signal, shutting down Serf agent", "signal", sig)
+
+		cancel()
+		ipc.Shutdown()
+		if err = serfAgent.Leave(); err != nil {
+			slog.Error("Gracefully leave Serf cluster", "error", err)
+		}
+		if err = serfAgent.Shutdown(); err != nil {
+			slog.Error("Gracefully shutdown Serf agent", "error", err)
+		}
+		fmt.Println("closing crdt store")
+		if err = store.Close(); err != nil {
+			slog.Error("Close CRDT store", "error", err)
+		}
+		fmt.Println("closing local store")
+		if err = localStore.Close(); err != nil {
+			slog.Error("Close local store", "error", err)
+		}
+
+		done <- true
+	}()
+
+	<-done
+}
@@ -0,0 +1,97 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/ipfs/boxo/datastore/dshelp"
+	dag "github.com/ipfs/boxo/ipld/merkledag"
+	"github.com/ipfs/go-cid"
+	ds "github.com/ipfs/go-datastore"
+	ipld "github.com/ipfs/go-ipld-format"
+	"log/slog"
+)
+
+// Implements the DAGService interface.
+// TODO: implement SessionDAGService to optimize node fetching.
+// TOOD: persistentSerfDAG?
+type dagSyncer struct {
+	// Persistent storage for the nodes.
+	store     ds.Datastore
+	namespace ds.Key
+}
+
+func newDAGSyncer(store ds.Datastore, namespace ds.Key) *dagSyncer {
+	return &dagSyncer{
+		store:     store,
+		namespace: namespace,
+	}
+}
+
+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))
+	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, 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 protoNode, nil
+}
+
+func (d *dagSyncer) GetMany(ctx context.Context, cids []cid.Cid) <-chan *ipld.NodeOption {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (d *dagSyncer) Add(ctx context.Context, node ipld.Node) error {
+	slog.Debug("Adding node", "cid", node.Cid())
+	protoNode, ok := node.(*dag.ProtoNode)
+	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.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
+}
+
+func (d *dagSyncer) AddMany(ctx context.Context, nodes []ipld.Node) error {
+	panic("implement me")
+}
+
+func (d *dagSyncer) Remove(ctx context.Context, cid cid.Cid) error {
+	panic("implement me")
+}
+
+func (d *dagSyncer) RemoveMany(ctx context.Context, cids []cid.Cid) error {
+	panic("implement me")
+}
+
+func (d *dagSyncer) nodeKey(cid cid.Cid) ds.Key {
+	return d.namespace.Child(dshelp.MultihashToDsKey(cid.Hash()))
+}
+
+var _ ipld.DAGService = (*dagSyncer)(nil)