Enable intial revision

This adds a flag which alters the initial state of the store. It
enables during operation to advance to a high rev on start to avoid
collisions with data imported/recovered.

doozerd.go

@@ -40,6 +40,7 @@ var (
 	hi          = flag.Int64("hist", 2000, "length of history/revisions to keep")
 	certFile    = flag.String("tlscert", "", "TLS public certificate")
 	keyFile     = flag.String("tlskey", "", "TLS private key")
+	rev         = flag.Int64("rev", 0, "Sets the initial rev the store starts at")
 )
 
 var (
@@ -134,7 +135,7 @@ func main() {
 		cl = boot(*name, id, *laddr, *buri)
 	}
 
-	peer.Main(*name, id, *buri, rwsk, rosk, cl, usock, tsock, wsock, ns(*pi), ns(*fd), ns(*kt), *hi)
+	peer.Main(*name, id, *buri, rwsk, rosk, cl, usock, tsock, wsock, ns(*pi), ns(*fd), ns(*kt), *hi, *rev)
 	panic("main exit")
 }
 

peer/peer.go

@@ -44,14 +44,20 @@ func (p *proposer) Propose(v []byte) (e store.Event) {
 	return
 }
 
-func Main(clusterName, self, buri, rwsk, rosk string, cl *doozer.Conn, udpConn *net.UDPConn, listener, webListener net.Listener, pulseInterval, fillDelay, kickTimeout int64, hi int64) {
+func Main(
+	clusterName, self, buri, rwsk, rosk string,
+	cl *doozer.Conn,
+	udpConn *net.UDPConn,
+	listener, webListener net.Listener,
+	pulseInterval, fillDelay, kickTimeout, hi, initialRev int64,
+) {
 	listenAddr := listener.Addr().String()
 
 	canWrite := make(chan bool, 1)
 	in := make(chan consensus.Packet, 50)
 	out := make(chan consensus.Packet, 50)
 
-	st := store.New()
+	st := store.New(initialRev)
 	pr := &proposer{
 		seqns: make(chan int64, alpha),
 		props: make(chan *consensus.Prop),

store/store.go

@@ -69,7 +69,7 @@ type state struct {
 // Creates a new, empty data store. Mutations will be applied in order,
 // starting at number 1 (number 0 can be thought of as the creation of the
 // store).
-func New() *Store {
+func New(initialRev int64) *Store {
 	ops := make(chan Op)
 	seqns := make(chan int64)
 	waiting := make(chan int)
@@ -81,7 +81,7 @@ func New() *Store {
 		watchCh:     make(chan *watch),
 		cancelWatch: make(chan (<-chan Event)),
 		watches:     watches{},
-		state:       &state{0, emptyDir},
+		state:       &state{initialRev, emptyDir},
 		log:         map[int64]Event{},
 		cleanCh:     make(chan int64),
 		flush:       make(chan bool),