func (b *bee) handoffNonPersistent(to uint64) error {
	s, err := b.stateL1.Save()
	if err != nil {
		return err
	}

	if _, err = b.qee.sendCmdToBee(to, cmdRestoreState{State: s}); err != nil {
		return err
	}

	oldc := b.colony()
	newc := oldc.DeepCopy()
	newc.Leader = to
	up := updateColony{
		Term: b.term(),
		Old:  oldc,
		New:  newc,
	}
	ctx, cnl := context.WithTimeout(context.Background(),
		10*b.hive.config.RaftElectTimeout())
	defer cnl()
	if _, err := b.hive.proposeAmongHives(ctx, up); err != nil {
		return err
	}

	b.becomeProxy()
	return nil
}

func (h *DeQHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	q, ok := mux.Vars(r)["queue"]
	if !ok {
		http.Error(w, "unkown queue", http.StatusBadRequest)
		return
	}

	ctx, cnl := context.WithTimeout(context.Background(), httpTimeout)
	defer cnl()

	d := Deque{Queue: Queue(q)}
	res, err := h.Hive.Sync(ctx, d)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	switch res := res.(type) {
	case Error:
		http.Error(w, res.Message, http.StatusInternalServerError)
	case Dequed:
		if err = json.NewEncoder(w).Encode(res.Task); err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
	}
}

func BenchmarkOneNode(b *testing.B) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	n := newNode()
	s := NewMemoryStorage()
	r := newTestRaft(1, []uint64{1}, 10, 1, s)
	go n.run(r)

	defer n.Stop()

	n.Campaign(ctx)
	go func() {
		for i := 0; i < b.N; i++ {
			n.Propose(ctx, []byte("foo"))
		}
	}()

	for {
		rd := <-n.Ready()
		s.Append(rd.Entries)
		// a reasonable disk sync latency
		time.Sleep(1 * time.Millisecond)
		n.Advance()
		if rd.HardState.Commit == uint64(b.N+1) {
			return
		}
	}
}

func TestNodeAdvance(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	storage := NewMemoryStorage()
	c := &Config{
		ID:              1,
		ElectionTick:    10,
		HeartbeatTick:   1,
		Storage:         storage,
		MaxSizePerMsg:   noLimit,
		MaxInflightMsgs: 256,
	}
	n := StartNode(c, []Peer{{ID: 1}})
	n.Campaign(ctx)
	<-n.Ready()
	n.Propose(ctx, []byte("foo"))
	var rd Ready
	select {
	case rd = <-n.Ready():
		t.Fatalf("unexpected Ready before Advance: %+v", rd)
	case <-time.After(time.Millisecond):
	}
	storage.Append(rd.Entries)
	n.Advance()
	select {
	case <-n.Ready():
	case <-time.After(time.Millisecond):
		t.Errorf("expect Ready after Advance, but there is no Ready available")
	}
}

// ProposeRetry proposes the request to the given group. It retires maxRetries
// times using the given timeout. If maxRetries is -1 it will keep proposing the
// request until it retrieves a response.
func (n *MultiNode) ProposeRetry(group uint64, req interface{},
	timeout time.Duration, maxRetries int) (res interface{}, err error) {

	for {
		ctx, ccl := context.WithTimeout(context.Background(), timeout)
		defer ccl()
		if status := n.Status(group); status == nil || status.SoftState.Lead == 0 {
			// wait with the hope that the group will be created.
			n.waitElection(ctx, group)
		} else {
			res, err = n.Propose(ctx, group, req)
			if err != context.DeadlineExceeded {
				return
			}
		}

		if maxRetries < 0 {
			continue
		}

		maxRetries--
		if maxRetries == 0 {
			break
		}
	}

	return nil, context.DeadlineExceeded
}

func (b *bee) addFollower(bid uint64, hid uint64) error {
	oldc := b.colony()
	if oldc.Leader != b.beeID {
		return fmt.Errorf("%v is not the leader", b)
	}
	newc := oldc.DeepCopy()
	if !newc.AddFollower(bid) {
		return ErrDuplicateBee
	}

	t := 10 * b.hive.config.RaftElectTimeout()
	upctx, upcnl := context.WithTimeout(context.Background(), t)
	defer upcnl()

	gid := oldc.ID
	// TODO(soheil): It's important to have a proper order here. Or launch both in
	// parallel and cancel them on error.
	up := updateColony{
		Term: b.term(),
		Old:  oldc,
		New:  newc,
	}
	if _, err := b.hive.proposeAmongHives(upctx, up); err != nil {
		glog.Errorf("%v cannot update its colony: %v", b, err)
		return err
	}

	cfgctx, cfgcnl := context.WithTimeout(context.Background(), t)
	defer cfgcnl()
	if err := b.hive.node.AddNodeToGroup(cfgctx, hid, gid, bid); err != nil {
		return err
	}

	cmd := cmd{
		Hive: hid,
		App:  b.app.Name(),
		Bee:  bid,
		Data: cmdJoinColony{Colony: newc},
	}
	if _, err := b.hive.client.sendCmd(cmd); err != nil {
		return err
	}

	b.setColony(newc)
	return nil
}

func TestSyncCancel(t *testing.T) {
	h := newHiveForTest()
	req := query("test")
	ctx, ccl := context.WithCancel(context.Background())
	go ccl()
	_, err := h.Sync(ctx, req)
	if err == nil {
		t.Errorf("no error in process: %v", err)
	}
}

func BenchmarkThroughput(b *testing.B) {
	hive := bh.NewHive()
	defer os.RemoveAll(hive.Config().StatePath)

	if b.N < 1024 {
		return
	}

	b.StopTimer()
	app := hive.NewApp("kvstore", bh.Persistent(1))
	store := &KVStore{
		Hive:    hive,
		Buckets: bees,
	}
	app.Handle(Put{}, store)
	app.Handle(Get(""), store)
	go hive.Start()

	keys := make([]string, 64)
	for i, _ := range keys {
		keys[i] = fmt.Sprintf("%dkeys%d", i, i)
	}
	val := "val"

	for _, k := range keys {
		hive.Emit(Put{Key: k, Val: val})
		hive.Sync(context.Background(), Get(k))
	}

	b.StartTimer()
	for i := 0; i < b.N; i++ {
		for _, k := range keys {
			hive.Emit(Put{Key: k, Val: val})
			i++
		}
	}

	for _, k := range keys {
		hive.Sync(context.Background(), Get(k))
	}
	b.StopTimer()
	hive.Stop()
}

// TestMultiNodeStart ensures that a node can be started correctly. The node should
// start with correct configuration change entries, and can accept and commit
// proposals.
func TestMultiNodeStart(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
	ccdata, err := cc.Marshal()
	if err != nil {
		t.Fatalf("unexpected marshal error: %v", err)
	}
	wants := []Ready{
		{
			SoftState: &SoftState{Lead: 1, RaftState: StateLeader},
			HardState: raftpb.HardState{Term: 2, Commit: 2, Vote: 1},
			Entries: []raftpb.Entry{
				{Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
				{Term: 2, Index: 2},
			},
			CommittedEntries: []raftpb.Entry{
				{Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
				{Term: 2, Index: 2},
			},
		},
		{
			HardState:        raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
			Entries:          []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
			CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
		},
	}
	mn := StartMultiNode(1)
	storage := NewMemoryStorage()
	mn.CreateGroup(1, newTestConfig(1, nil, 10, 1, storage), []Peer{{ID: 1}})
	mn.Campaign(ctx, 1)
	gs := <-mn.Ready()
	g := gs[1]
	if !reflect.DeepEqual(g, wants[0]) {
		t.Fatalf("#%d: g = %+v,\n             w   %+v", 1, g, wants[0])
	} else {
		storage.Append(g.Entries)
		mn.Advance(gs)
	}

	mn.Propose(ctx, 1, []byte("foo"))
	if gs2 := <-mn.Ready(); !reflect.DeepEqual(gs2[1], wants[1]) {
		t.Errorf("#%d: g = %+v,\n             w   %+v", 2, gs2[1], wants[1])
	} else {
		storage.Append(gs2[1].Entries)
		mn.Advance(gs2)
	}

	select {
	case rd := <-mn.Ready():
		t.Errorf("unexpected Ready: %+v", rd)
	case <-time.After(time.Millisecond):
	}
}

func TestSyncDeferReply(t *testing.T) {
	h := newHiveForTest()
	app := h.NewApp("syncDeferReply")
	type reply string
	type deferred struct {
		Repliable
		Q query
	}
	deferredCh := make(chan struct{})

	deferf := func(msg Msg, ctx RcvContext) error {
		deferredCh <- struct{}{}
		r := ctx.DeferReply(msg)
		return ctx.Dict("sync").Put("reply", deferred{
			Repliable: r,
			Q:         msg.Data().(query),
		})
	}

	replyf := func(msg Msg, ctx RcvContext) error {
		v, err := ctx.Dict("sync").Get("reply")
		if err != nil {
			t.Fatalf("cannot decode reply: %v", err)
		}
		d := v.(deferred)
		d.Reply(ctx, d.Q)
		return nil
	}

	mapf := func(msg Msg, ctx MapContext) MappedCells {
		return ctx.LocalMappedCells()
	}

	app.HandleFunc(query(""), mapf, deferf)
	app.HandleFunc(reply(""), mapf, replyf)

	go h.Start()
	defer h.Stop()

	go func() {
		<-deferredCh
		h.Emit(reply(""))
	}()

	req := query("test")
	res, err := h.Sync(context.Background(), req)
	if err != nil {
		t.Fatalf("error in process: %v", err)
	}
	if res != req {
		t.Errorf("sync.Process(%v) = %v; want=%v", req, res, req)
	}
}

func (s *rpcServer) ProcessRaft(batch raft.Batch, dummy *bool) (err error) {
	if batch.To != s.h.ID() {
		glog.Fatalf("%v recieves a raft message for %v", s.h, msg.To)
	}

	glog.V(3).Infof("%v handles a batch from %v", s.h, batch.From)
	ctx, cnl := context.WithTimeout(context.Background(),
		s.h.config.RaftHBTimeout())
	err = s.h.node.StepBatch(ctx, batch, 2*s.h.config.RaftHBTimeout())
	cnl()
	return
}

func ExampleWithTimeout() {
	// Pass a context with a timeout to tell a blocking function that it
	// should abandon its work after the timeout elapses.
	ctx, _ := context.WithTimeout(context.Background(), 100*time.Millisecond)
	select {
	case <-time.After(200 * time.Millisecond):
		fmt.Println("overslept")
	case <-ctx.Done():
		fmt.Println(ctx.Err()) // prints "context deadline exceeded"
	}
	// Output:
	// context deadline exceeded
}

func (b *bee) replicate() error {
	glog.V(2).Infof("%v replicates transaction", b)
	b.Lock()

	if b.stateL2 != nil {
		err := b.commitTxL2()
		b.stateL2 = nil
		if err != nil && err != state.ErrNoTx {
			b.Unlock()
			return err
		}
	}

	if b.stateL1.TxStatus() != state.TxOpen {
		b.Unlock()
		return state.ErrNoTx
	}

	stx := b.stateL1.Tx()
	if len(stx.Ops) == 0 {
		err := b.commitTxL1()
		b.Unlock()
		return err
	}

	b.Unlock()

	if err := b.maybeRecruitFollowers(); err != nil {
		return err
	}

	msgs := make([]*msg, len(b.msgBufL1))
	copy(msgs, b.msgBufL1)
	tx := tx{
		Tx:   stx,
		Msgs: msgs,
	}
	ctx, cnl := context.WithTimeout(context.Background(),
		10*b.hive.config.RaftElectTimeout())
	defer cnl()
	commit := commitTx{
		Tx:   tx,
		Term: b.term(),
	}
	if _, err := b.hive.node.Propose(ctx, b.group(), commit); err != nil {
		glog.Errorf("%v cannot replicate the transaction: %v", b, err)
		return err
	}
	glog.V(2).Infof("%v successfully replicates transaction", b)
	return nil
}

func (s *KVStore) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	k, ok := mux.Vars(r)["key"]
	if !ok {
		http.Error(w, "no key in the url", http.StatusBadRequest)
		return
	}

	ctx, cnl := context.WithTimeout(context.Background(), 30*time.Second)
	var res interface{}
	var err error
	switch r.Method {
	case "GET":
		res, err = s.Hive.Sync(ctx, Get(k))
	case "PUT":
		var v []byte
		v, err = ioutil.ReadAll(r.Body)
		if err != nil {
			break
		}
		res, err = s.Hive.Sync(ctx, Put{Key: k, Val: string(v)})
	case "DELETE":
		res, err = s.Hive.Sync(ctx, Del(k))
	}
	cnl()

	if err != nil {
		switch {
		case err.Error() == errKeyNotFound.Error():
			http.Error(w, err.Error(), http.StatusNotFound)
			return
		case err.Error() == errInternal.Error():
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		case err.Error() == errInvalid.Error():
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
	}

	if res == nil {
		return
	}

	js, err := json.Marshal(res)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Write(js)
}

func BenchmarkSync(b *testing.B) {
	log.SetOutput(ioutil.Discard)
	b.StopTimer()

	h := newHiveForTest()
	app := h.NewApp("sync")
	app.Handle(0, benchSyncHandler{})

	go h.Start()
	defer h.Stop()
	waitTilStareted(h)

	b.StartTimer()
	for i := 0; i < b.N; i++ {
		h.Sync(context.Background(), i)
	}
	b.StopTimer()
}

func (n *MultiNode) handleBatch(bt batchTimeout) {
	ctx, cnl := context.WithTimeout(context.Background(), bt.timeout)
	for g, msgs := range bt.batch.Messages {
		if _, ok := n.groups[g]; !ok {
			glog.Errorf("group %v is not created on %v", g, n)
			continue
		}
		for _, m := range msgs {
			if err := n.node.Step(ctx, g, m); err != nil {
				glog.Errorf("%v cannot step group %v: %v", n, g, err)
				if err == context.DeadlineExceeded || err == context.Canceled {
					return
				}
			}
		}
	}
	cnl()
}

// Cancel and Stop should unblock Step()
func TestNodeStepUnblock(t *testing.T) {
	// a node without buffer to block step
	n := &node{
		propc: make(chan raftpb.Message),
		done:  make(chan struct{}),
	}

	ctx, cancel := context.WithCancel(context.Background())
	stopFunc := func() { close(n.done) }

	tests := []struct {
		unblock func()
		werr    error
	}{
		{stopFunc, ErrStopped},
		{cancel, context.Canceled},
	}

	for i, tt := range tests {
		errc := make(chan error, 1)
		go func() {
			err := n.Step(ctx, raftpb.Message{Type: raftpb.MsgProp})
			errc <- err
		}()
		tt.unblock()
		select {
		case err := <-errc:
			if err != tt.werr {
				t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
			}
			//clean up side-effect
			if ctx.Err() != nil {
				ctx = context.TODO()
			}
			select {
			case <-n.done:
				n.done = make(chan struct{})
			default:
			}
		case <-time.After(time.Millisecond * 100):
			t.Errorf("#%d: failed to unblock step", i)
		}
	}
}

func TestMultiNodeAdvance(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	storage := NewMemoryStorage()
	mn := StartMultiNode(1)
	mn.CreateGroup(1, newTestConfig(1, nil, 10, 1, storage), []Peer{{ID: 1}})
	mn.Campaign(ctx, 1)
	rd1 := <-mn.Ready()
	mn.Propose(ctx, 1, []byte("foo"))
	select {
	case rd2 := <-mn.Ready():
		t.Fatalf("unexpected Ready before Advance: %+v", rd2)
	case <-time.After(time.Millisecond):
	}
	storage.Append(rd1[1].Entries)
	mn.Advance(rd1)
	select {
	case <-mn.Ready():
	case <-time.After(time.Millisecond):
		t.Errorf("expect Ready after Advance, but there is no Ready available")
	}
}

func ExampleSyncInstall() {
	rcvf := func(msg Msg, ctx RcvContext) error {
		ctx.Reply(msg, "hello "+msg.Data().(query))
		return nil
	}
	mapf := func(msg Msg, ctx MapContext) MappedCells {
		return ctx.LocalMappedCells()
	}

	hive := NewHive()
	app := hive.NewApp("sync-app")
	app.HandleFunc(query(""), mapf, rcvf)

	go hive.Start()
	defer hive.Stop()

	result, err := hive.Sync(context.Background(), query("your name"))
	if err != nil {
		fmt.Printf("error in sync: %v", err)
		return
	}
	fmt.Println(result)
}

func TestSync(t *testing.T) {
	h := newHiveForTest()
	app := h.NewApp("sync")
	rcvf := func(msg Msg, ctx RcvContext) error {
		ctx.Reply(msg, msg.Data().(query))
		return nil
	}
	mapf := func(msg Msg, ctx MapContext) MappedCells {
		return ctx.LocalMappedCells()
	}
	app.HandleFunc(query(""), mapf, rcvf)

	go h.Start()
	defer h.Stop()

	req := query("test")
	res, err := h.Sync(context.Background(), req)
	if err != nil {
		t.Fatalf("error in process: %v", err)
	}
	if res != req {
		t.Errorf("sync.Process(%v) = %v; want=%v", req, res, req)
	}
}