func (n *Node) Update(meta *transport.NodeMeta) error {
	n.log.Printf("Received metadata update: %+v\n", meta)
	n.mu.Lock()
	defer n.mu.Unlock()
	n.IsHead = meta.IsHead
	n.IsTail = meta.IsTail
	if err := n.connectToPredecessor(meta.Prev); err != nil {
		return err
	}
	tail := n.neighbors[transport.NeighborPosTail]
	if !meta.IsTail && tail.address != meta.Tail && meta.Tail != "" {
		err := n.connectToNode(meta.Tail, transport.NeighborPosTail)
		if err != nil {
			return err
		}
	}
	if err := n.connectToSuccessor(meta.Next); err != nil {
		return err
	}
	if meta.IsTail {
		dirty, err := n.store.AllDirty()
		if err != nil {
			n.log.Println("Error fetching all dirty items during node Update")
			return err
		}
		for i := range dirty {
			go func(item *store.Item) {
				if err := n.commitAndSend(item.Key, item.Version); err != nil {
					n.log.Printf(
						"Error during commit & send for item: %#v, error: %#v\n",
						item,
						err,
					)
				}
			}(dirty[i])
		}
	}
	return nil
}
Update is for updating a node's metadata. If new neighbors are given, the
Node will disconnect from the current neighbors before connecting to the new
ones. Coordinator uses this method to update metadata of the node when there
is a failure or re-organization of the chain.
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
}
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
}
func (n *Node) requestFwdPropagation(client transport.NodeClient) error {
	dirty, err := n.store.AllDirty()
	if err != nil {
		n.log.Printf("Failed to get all dirty items: %#v\n", err)
		return err
	}
	reply, err := client.FwdPropagate(propagateRequestFromItems(dirty))
	if err != nil {
		n.log.Printf("Failed during forward propagation: %#v\n", err)
		return err
	}
	return n.writePropagated(reply)
}
func (n *Node) requestBackPropagation(client transport.NodeClient) error {
	committed, err := n.store.AllCommitted()
	if err != nil {
		n.log.Printf("Failed to get all committed items: %#v\n", err)
		return err
	}
	reply, err := client.BackPropagate(propagateRequestFromItems(committed))
	if err != nil {
		n.log.Printf("Failed during back propagation: %#v\n", err)
		return err
	}
	return n.commitPropagated(reply)
}
func (n *Node) resetNeighbor(pos transport.NeighborPos) {
	n.neighbors[pos].rpc.Close()
	n.neighbors[pos] = neighbor{}
}
func (n *Node) Ping() error {
	return nil
}
func (n *Node) connectToPredecessor(address string) error {
	prev := n.neighbors[transport.NeighborPosPrev]
	if prev.address == address {
		n.log.Println("New predecessor same address as last one, keeping conn.")
		return nil
	} else if address == "" {
		n.log.Println("Resetting predecessor")
		n.resetNeighbor(transport.NeighborPosPrev)
		return nil
	}
	n.log.Printf("connecting to new predecessor %s\n", address)
	if err := n.connectToNode(address, transport.NeighborPosPrev); err != nil {
		return err
	}
	prevC := n.neighbors[transport.NeighborPosPrev].rpc
	return n.requestFwdPropagation(prevC)
}
func (n *Node) connectToSuccessor(address string) error {
	next := n.neighbors[transport.NeighborPosNext]
	if next.address == address {
		n.log.Println("New successor same address as last one, keeping conn.")
		return nil
	} else if address == "" {
		n.log.Println("Resetting successor")
		n.resetNeighbor(transport.NeighborPosNext)
		return nil
	}
	n.log.Printf("connecting to new successor %s\n", address)
	if err := n.connectToNode(address, transport.NeighborPosNext); err != nil {
		return err
	}
	nextC := n.neighbors[transport.NeighborPosNext].rpc
	return n.requestBackPropagation(nextC)
}