catch up

Author: xDarksome

ballot_protocol.go

@@ -7,8 +7,8 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-type Catchuper interface {
-	GetSlot(slotIndex uint64, nodeID string) (Slot, error)
+type SlotsLoader interface {
+	LoadSlots(from uint64, to Slot) []Slot
 }
 
 type Ledger interface {
@@ -19,6 +19,7 @@ type ballotProtocol struct {
 	slotIndex uint64
 	id        string
 
+	loader SlotsLoader
 	ledger Ledger
 
 	quorumSlices
@@ -36,17 +37,44 @@ type ballotProtocol struct {
 	nominationProtocol chan protocolMessage
 	lastCandidate      Value
 	candidates         chan Value
+
+	catchUpDone  chan struct{}
+	persistLater []Slot
+}
+
+func (b *ballotProtocol) init(slotIndex uint64) {
+	b.slotIndex = slotIndex
+	b.ballots = newBallots()
+	b.currentBallot = &ballot{slotIndex: slotIndex, counter: 1}
+	b.counters = make(ballotCounters)
+	b.timer = time.NewTimer(time.Hour)
+}
+
+func (b *ballotProtocol) reinit(index uint64) {
+	b.init(index)
+
+	b.candidates = make(chan Value, 1000)
+	b.highestAcceptedPrepared = nil
+	b.highestConfirmedPrepared = nil
+	b.lastCandidate = nil
+
+	b.nominationProtocol <- protocolMessage{
+		slotIndex:  index,
+		candidates: b.candidates,
+	}
 }
 
 func (b *ballotProtocol) run() {
 	for {
 		select {
-		case <-b.timer.C:
-			b.updateCurrentBallotCounter(b.currentBallot.counter + 1)
 		case c := <-b.candidates:
 			b.newCandidate(c)
 		case m := <-b.inputMessages:
 			b.receive(m)
+		case <-b.timer.C:
+			b.updateCurrentBallotCounter(b.currentBallot.counter + 1)
+		case <-b.catchUpDone:
+			b.finishCatchUp()
 		}
 	}
 }
@@ -80,6 +108,12 @@ func (b *ballotProtocol) receive(m *Message) {
 	}
 }
 
+func (b *ballotProtocol) updateCurrentBallotCounter(c uint32) {
+	b.currentBallot.counter = c
+	b.recomputeCurrentBallotValue()
+	b.votePrepare()
+}
+
 func (b *ballotProtocol) broadcast(m *Message) {
 	if m.SlotIndex != b.slotIndex {
 		return
@@ -235,37 +269,41 @@ func (b *ballotProtocol) confirmCommit(ballot *ballot) {
 }
 
 func (b *ballotProtocol) externalize(s Slot) {
+	if s.Index != b.slotIndex {
+		b.catchUp(s)
+		b.reinit(s.Index)
+		return
+	}
+
 	b.ledger.PersistSlot(s)
 	fmt.Println("\n", b.id, "externalized", s.Index, string(s.Value))
 	b.reinit(s.Index + 1)
 }
 
-func (b *ballotProtocol) init(slotIndex uint64) {
-	b.slotIndex = slotIndex
-	b.ballots = newBallots()
-	b.currentBallot = &ballot{slotIndex: slotIndex, counter: 1}
-	b.counters = make(ballotCounters)
-	b.timer = time.NewTimer(time.Hour)
-}
+func (b *ballotProtocol) catchUp(slot Slot) {
+	inProgress := b.persistLater != nil
+	b.persistLater = append(b.persistLater, slot)
 
-func (b *ballotProtocol) reinit(index uint64) {
-	b.init(index)
+	if inProgress {
+		return
+	}
 
-	b.candidates = make(chan Value, 1000)
-	b.highestAcceptedPrepared = nil
-	b.highestConfirmedPrepared = nil
-	b.lastCandidate = nil
+	go catchuper{
+		loader: b.loader,
+		ledger: b.ledger,
+		from:   b.slotIndex,
+		to:     slot,
+		done:   b.catchUpDone,
+	}.catchUp()
+}
 
-	b.nominationProtocol <- protocolMessage{
-		slotIndex:  index,
-		candidates: b.candidates,
+func (b *ballotProtocol) finishCatchUp() {
+	for _, slot := range b.persistLater {
+		b.ledger.PersistSlot(slot)
 	}
-}
 
-func (b *ballotProtocol) updateCurrentBallotCounter(c uint32) {
-	b.currentBallot.counter = c
-	b.recomputeCurrentBallotValue()
-	b.votePrepare()
+	b.slotIndex = b.persistLater[len(b.persistLater)-1].Index + 1
+	b.persistLater = nil
 }
 
 func (b *ballotProtocol) recomputeCurrentBallotValue() {

catchup.go

@@ -0,0 +1,17 @@
+package scp
+
+type catchuper struct {
+	loader SlotsLoader
+	ledger Ledger
+	from   uint64
+	to     Slot
+	done   chan struct{}
+}
+
+func (c catchuper) catchUp() {
+	slots := c.loader.LoadSlots(c.from, c.to)
+	for _, s := range slots {
+		c.ledger.PersistSlot(s)
+	}
+	c.done <- struct{}{}
+}

main.go

@@ -15,6 +15,7 @@ type Config struct {
 	Validator    Validator
 	Combiner     Combiner
 	Ledger       Ledger
+	SlotsLoader  SlotsLoader
 	QuorumSlices []*QuorumSlice
 }
 
@@ -47,11 +48,13 @@ func New(cfg Config) *Consensus {
 		slotIndex:          cfg.CurrentSlot,
 		id:                 cfg.NodeID,
 		ledger:             cfg.Ledger,
+		loader:             cfg.SlotsLoader,
 		quorumSlices:       cfg.QuorumSlices,
 		inputMessages:      make(chan *Message, 1000000),
 		outputMessages:     outputMessages,
 		candidates:         candidates,
 		nominationProtocol: link,
+		catchUpDone:        make(chan struct{}),
 	}
 	ballotProtocol.init(cfg.CurrentSlot)
 

main_test.go

@@ -10,7 +10,21 @@ import (
 	"github.com/xdarksome/scp"
 )
 
-type app struct{}
+type app struct {
+	slots map[uint64]scp.Slot
+}
+
+func newApp() app {
+	return app{
+		slots: map[uint64]scp.Slot{
+			1: {Index: 1, Value: []byte{1}},
+			2: {Index: 2, Value: []byte{2}},
+			3: {Index: 3, Value: []byte{3}},
+			4: {Index: 4, Value: []byte{4}},
+		},
+	}
+}
+
 type badApp struct{}
 
 func (v app) ValidateValue(scp.Value) bool {
@@ -29,7 +43,21 @@ func (v app) CombineValues(values ...scp.Value) (composite scp.Value) {
 	return composite
 }
 
-func (v app) PersistSlot(slot scp.Slot) {}
+func (v app) PersistSlot(s scp.Slot) {
+	v.slots[s.Index] = s
+}
+
+type catchup struct {
+	app
+}
+
+func (c catchup) LoadSlots(from uint64, to scp.Slot) (loaded []scp.Slot) {
+	time.Sleep(5 * time.Second)
+	for i := from; i < to.Index; i++ {
+		loaded = append(loaded, c.slots[i])
+	}
+	return loaded
+}
 
 func (v badApp) ValidateValue(segment scp.Value) bool {
 	return false
@@ -49,10 +77,10 @@ func Test(t *testing.T) {
 
 	node1 := scp.New(scp.Config{
 		NodeID:      node1key,
-		CurrentSlot: 10,
-		Validator:   app{},
-		Combiner:    app{},
-		Ledger:      app{},
+		CurrentSlot: 5,
+		Validator:   newApp(),
+		Combiner:    newApp(),
+		Ledger:      newApp(),
 		QuorumSlices: []*scp.QuorumSlice{
 			{
 				Threshold: 2,
@@ -67,10 +95,10 @@ func Test(t *testing.T) {
 
 	node2 := scp.New(scp.Config{
 		NodeID:      node2key,
-		CurrentSlot: 10,
-		Validator:   app{},
-		Combiner:    app{},
-		Ledger:      app{},
+		CurrentSlot: 5,
+		Validator:   newApp(),
+		Combiner:    newApp(),
+		Ledger:      newApp(),
 		QuorumSlices: []*scp.QuorumSlice{
 			{
 				Threshold: 2,
@@ -83,12 +111,13 @@ func Test(t *testing.T) {
 		},
 	})
 
+	node3app := newApp()
 	node3 := scp.New(scp.Config{
 		NodeID:      node3key,
-		CurrentSlot: 10,
-		Validator:   app{},
-		Combiner:    app{},
-		Ledger:      app{},
+		CurrentSlot: 5,
+		Validator:   node3app,
+		Combiner:    node3app,
+		Ledger:      node3app,
 		QuorumSlices: []*scp.QuorumSlice{
 			{
 				Threshold: 2,
@@ -103,10 +132,11 @@ func Test(t *testing.T) {
 
 	node4 := scp.New(scp.Config{
 		NodeID:      node4key,
-		CurrentSlot: 10,
-		Validator:   badApp{},
-		Combiner:    badApp{},
-		Ledger:      badApp{},
+		CurrentSlot: 1,
+		Validator:   newApp(),
+		Combiner:    newApp(),
+		Ledger:      newApp(),
+		SlotsLoader: &catchup{node3app},
 		QuorumSlices: []*scp.QuorumSlice{
 			{
 				Threshold: 2,