func propagateRequestFromItems(items []*store.Item) *transport.PropagateRequest {
	req := transport.PropagateRequest{}
	for _, item := range items {
		if req[item.Key] < item.Version {
			req[item.Key] = item.Version
		}
	}
	return &req
}
Create a map of the highest versions for each key.
package node
import (
	"errors"
	"log"
	"sync"
	"github.com/despreston/go-craq/store"
	"github.com/despreston/go-craq/transport"
)
type neighbor struct {
	rpc     transport.NodeClient
	address string
}
type Opts struct {
	Store store.Storer
	Address string
	PubAddress string
	CdrAddress string
	Transport transport.NodeClientFactory
	CoordinatorClient transport.CoordinatorClient
	Log *log.Logger
}
type commitEvent struct {
	Key     string
	Version uint64
}
type Node struct {
	neighbors map[transport.NeighborPos]neighbor
	store store.Storer
	latest map[string]uint64
	committed                    chan commitEvent
	cdrAddress, address, pubAddr string
	cdr                          transport.CoordinatorClient
	IsHead, IsTail               bool
	mu                           sync.Mutex
	transport                    func() transport.NodeClient
	log                          *log.Logger
}
func New(opts Opts) *Node {
	logger := opts.Log
	if opts.Log == nil {
		logger = log.Default()
	}
	return &Node{
		latest:     make(map[string]uint64),
		neighbors:  make(map[transport.NeighborPos]neighbor, 3),
		cdrAddress: opts.CdrAddress,
		address:    opts.Address,
		store:      opts.Store,
		transport:  opts.Transport,
		pubAddr:    opts.PubAddress,
		cdr:        opts.CoordinatorClient,
		log:        logger,
	}
}
func (n *Node) Start() error {
	if err := n.backfillLatest(); err != nil {
		log.Fatalf("Failed to backfill latest versions.\n Error: %#v", err)
	}
	if err := n.connectToCoordinator(); err != nil {
		log.Fatalf("Failed to connect to the chain.\n Error: %#v", err)
	}
	return nil
}
func (n *Node) backfillLatest() error {
	c, err := n.store.AllCommitted()
	if err != nil {
		return err
	}
	for _, item := range c {
		n.latest[item.Key] = item.Version
	}
	return nil
}
func (n *Node) connectToCoordinator() error {
	err := n.cdr.Connect(n.cdrAddress)
	if err != nil {
		n.log.Println("Error connecting to the coordinator")
		return err
	}
	n.log.Printf("Connected to coordinator at %s\n", n.cdrAddress)
	reply, err := n.cdr.AddNode(n.pubAddr)
	if err != nil {
		n.log.Println(err.Error())
		return err
	}
	n.IsHead = reply.IsHead
	n.IsTail = reply.IsTail
	n.neighbors[transport.NeighborPosTail] = neighbor{address: reply.Tail}
	if reply.Prev != "" {
		if err := n.connectToNode(reply.Prev, transport.NeighborPosPrev); err != nil {
			n.log.Printf("Failed to connect to node in ConnectToCoordinator. %v\n", err)
			return err
		}
		if err := n.fullPropagate(); err != nil {
			return err
		}
	} else if n.neighbors[transport.NeighborPosPrev].address != "" {
		n.neighbors[transport.NeighborPosPrev].rpc.Close()
	}
	return nil
}
func (n *Node) fullPropagate() error {
	prevNeighbor := n.neighbors[transport.NeighborPosPrev].rpc
	if err := n.requestFwdPropagation(prevNeighbor); err != nil {
		return err
	}
	return n.requestBackPropagation(prevNeighbor)
}
func (n *Node) connectToNode(address string, pos transport.NeighborPos) error {
	newNbr := n.transport()
	if err := newNbr.Connect(address); err != nil {
		return err
	}
	n.log.Printf("connected to %s\n", address)
	nbr := n.neighbors[pos]
	if nbr.rpc != nil {
		nbr.rpc.Close()
	}
	n.neighbors[pos] = neighbor{
		rpc:     newNbr,
		address: address,
	}
	return nil
}
func (n *Node) writePropagated(reply *transport.PropagateResponse) error {
	for key, forKey := range *reply {
		for _, item := range forKey {
			if err := n.store.Write(key, item.Value, item.Version); err != nil {
				n.log.Printf("Failed to write item %+v to store: %#v\n", item, err)
				return err
			}
			n.log.Printf("wrote %s", key)
		}
	}
	return nil
}
func (n *Node) commit(key string, version uint64) error {
	if err := n.store.Commit(key, version); err != nil {
		n.log.Printf("Failed to commit. Key: %s Version: %d Error: %#v", key, version, err)
		return err
	}
	n.latest[key] = version
	if n.committed != nil {
		n.committed <- commitEvent{Key: key, Version: version}
	}
	return nil
}
func (n *Node) commitPropagated(reply *transport.PropagateResponse) error {
	for key, forKey := range *reply {
		for _, item := range forKey {
			if err := n.commit(key, item.Version); err != nil {
				if err == store.ErrNotFound {
					if err := n.store.Write(key, item.Value, item.Version); err != nil {
						return err
					}
					if err := n.commit(key, item.Version); err != nil {
						return err
					}
				} else {
					return err
				}
			}
		}
	}
	return nil
}