[Merged by Bors] - recover only N latest layers required for correct ballots decoding

node/node.go

@@ -625,7 +625,7 @@ func (app *App) initServices(ctx context.Context) error {
 	trtl, err := tortoise.Recover(
 		ctx,
 		app.cachedDB,
-		app.clock.CurrentLayer(), beaconProtocol, trtlopts...,
+		app.clock.CurrentLayer(), trtlopts...,
 	)
 	if err != nil {
 		return fmt.Errorf("can't recover tortoise state: %w", err)

tortoise/algorithm.go

@@ -114,6 +114,24 @@
 	return t, nil
 }
 
+func (t *Tortoise) RecoverFrom(lid types.LayerID, opinion, prev types.Hash32) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.logger.Debug("recover from",
+		zap.Uint32("lid", lid.Uint32()),
+		log.ZShortStringer("opinion", opinion),
+		log.ZShortStringer("prev opinion", prev),
+	)
+	t.trtl.evicted = lid - 1
+	t.trtl.pending = lid
+	t.trtl.verified = lid
+	t.trtl.processed = lid
+	t.trtl.last = lid
+	layer := t.trtl.layer(lid)
+	layer.opinion = opinion
+	layer.prevOpinion = &prev
+}
+
 // LatestComplete returns the latest verified layer.
 func (t *Tortoise) LatestComplete() types.LayerID {
 	t.mu.Lock()
@@ -300,6 +318,21 @@
 	if t.tracer != nil {
 		t.tracer.On(&BallotTrace{Ballot: ballot})
 	}
+}
+
+// OnRecoveredBallot is called for ballots recovered from database.
+//
+// For recovered ballots base ballot is not required to be in state therefore
+// opinion is not recomputed, but instead recovered from database state.
+func (t *Tortoise) OnRecoveredBallot(ballot *types.BallotTortoiseData) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if err := t.trtl.onRecoveredBallot(ballot); err != nil {
+		errorsCounter.Inc()
+		t.logger.Error("failed to save state from recovered ballot",
+			zap.Stringer("ballot", ballot.ID),
+			zap.Error(err))
+	}
 	if t.tracer != nil {
 		t.tracer.On(&BallotTrace{Ballot: ballot})
 	}
@@ -536,6 +569,11 @@
 // pending layer to the layer above equal layer.
 // this method is meant to be used only in recovery from disk codepath.
 func (t *Tortoise) resetPending(lid types.LayerID, opinion types.Hash32) bool {
+	t.logger.Debug("reset pending",
+		zap.Uint32("lid", lid.Uint32()),
+		log.ZShortStringer("computed", t.trtl.layer(lid).opinion),
+		log.ZShortStringer("stored", opinion),
+	)
 	if t.trtl.layer(lid).opinion == opinion {
 		t.trtl.pending = lid + 1
 		return true

tortoise/model/core.go

@@ -14,6 +14,7 @@
 	"github.com/spacemeshos/go-spacemesh/sql"
 	"github.com/spacemeshos/go-spacemesh/sql/atxs"
 	"github.com/spacemeshos/go-spacemesh/sql/ballots"
+	"github.com/spacemeshos/go-spacemesh/sql/beacons"
 	"github.com/spacemeshos/go-spacemesh/sql/blocks"
 	"github.com/spacemeshos/go-spacemesh/sql/certificates"
 	"github.com/spacemeshos/go-spacemesh/sql/layers"
@@ -32,13 +33,12 @@
 		panic(err)
 	}
 	c := &core{
-		id:      id,
-		logger:  logger,
-		rng:     rng,
-		cdb:     cdb,
-		beacons: newBeaconStore(),
-		units:   units,
-		signer:  sig,
+		id:     id,
+		logger: logger,
+		rng:    rng,
+		cdb:    cdb,
+		units:  units,
+		signer: sig,
 	}
 	cfg := tortoise.DefaultConfig()
 	cfg.LayerSize = layerSize
@@ -59,7 +59,6 @@
 	rng    *rand.Rand
 
 	cdb      *datastore.CachedDB
-	beacons  *beaconStore
 	tortoise *tortoise.Tortoise
 
 	// generated on setup
@@ -107,11 +106,11 @@
 		if c.refBallot != nil {
 			ballot.RefBallot = *c.refBallot
 		} else {
-			beacon, err := c.beacons.GetBeacon(ev.LayerID.GetEpoch())
+			beacon, err := beacons.Get(c.cdb, ev.LayerID.GetEpoch())
 			if err != nil {
 				beacon = types.Beacon{}
 				c.rng.Read(beacon[:])
-				c.beacons.StoreBeacon(ev.LayerID.GetEpoch(), beacon)
+				beacons.Set(c.cdb, ev.LayerID.GetEpoch(), beacon)
 			}
 			ballot.EpochData = &types.EpochData{
 				ActiveSetHash:    types.Hash32{1, 2, 3},
@@ -129,7 +128,8 @@
 		m.Send(MessageBallot{Ballot: ballot})
 	case MessageLayerEnd:
 		if ev.LayerID.After(types.GetEffectiveGenesis()) {
-			tortoise.RecoverLayer(context.Background(), c.tortoise, c.cdb, c.beacons, ev.LayerID, ev.LayerID, ev.LayerID, ev.LayerID)
+			tortoise.RecoverLayer(context.Background(), c.tortoise, c.cdb, ev.LayerID, c.tortoise.OnBallot)
+			c.tortoise.TallyVotes(context.Background(), ev.LayerID)
 			m.Notify(EventVerified{ID: c.id, Verified: c.tortoise.LatestComplete(), Layer: ev.LayerID})
 		}
 
@@ -172,30 +172,8 @@
 		}
 		atxs.Add(c.cdb, vAtx)
 	case MessageBeacon:
-		c.beacons.StoreBeacon(ev.EpochID, ev.Beacon)
+		beacons.Add(c.cdb, ev.EpochID+1, ev.Beacon)
 	case MessageCoinflip:
 		layers.SetWeakCoin(c.cdb, ev.LayerID, ev.Coinflip)
 	}
 }
-
-func newBeaconStore() *beaconStore {
-	return &beaconStore{
-		beacons: map[types.EpochID]types.Beacon{},
-	}
-}
-
-type beaconStore struct {
-	beacons map[types.EpochID]types.Beacon
-}
-
-func (b *beaconStore) GetBeacon(eid types.EpochID) (types.Beacon, error) {
-	beacon, exist := b.beacons[eid-1]
-	if !exist {
-		return types.Beacon{}, sql.ErrNotFound
-	}
-	return beacon, nil
-}
-
-func (b *beaconStore) StoreBeacon(eid types.EpochID, beacon types.Beacon) {
-	b.beacons[eid] = beacon
-}

tortoise/recover.go

@@ -10,15 +10,15 @@
 	"github.com/spacemeshos/go-spacemesh/sql"
 	"github.com/spacemeshos/go-spacemesh/sql/atxs"
 	"github.com/spacemeshos/go-spacemesh/sql/ballots"
+	"github.com/spacemeshos/go-spacemesh/sql/beacons"
 	"github.com/spacemeshos/go-spacemesh/sql/blocks"
 	"github.com/spacemeshos/go-spacemesh/sql/certificates"
 	"github.com/spacemeshos/go-spacemesh/sql/identities"
 	"github.com/spacemeshos/go-spacemesh/sql/layers"
-	"github.com/spacemeshos/go-spacemesh/system"
 )
 
 // Recover tortoise state from database.
-func Recover(ctx context.Context, db *datastore.CachedDB, latest types.LayerID, beacon system.BeaconGetter, opts ...Opt) (*Tortoise, error) {
+func Recover(ctx context.Context, db *datastore.CachedDB, current types.LayerID, opts ...Opt) (*Tortoise, error) {
 	trtl, err := New(opts...)
 	if err != nil {
 		return nil, err
@@ -29,6 +29,25 @@
 		return nil, fmt.Errorf("failed to load latest known layer: %w", err)
 	}
 
+	applied, err := layers.GetLastApplied(db)
+	if err != nil {
+		return nil, fmt.Errorf("get last applied: %w", err)
+	}
+	start := types.GetEffectiveGenesis() + 1
+	if applied > types.LayerID(trtl.cfg.WindowSize) {
+		window := applied - types.LayerID(trtl.cfg.WindowSize)
+		window = window.GetEpoch().FirstLayer() // windback to the start of the epoch to load ref ballots
+		if window > start {
+			prev, err1 := layers.GetAggregatedHash(db, window-1)
+			opinion, err2 := layers.GetAggregatedHash(db, window)
+			if err1 == nil && err2 == nil {
+				// tortoise will need reference to previous layer
+				trtl.RecoverFrom(window, opinion, prev)
+				start = window
+			}
+		}
+	}
+
 	malicious, err := identities.GetMalicious(db)
 	if err != nil {
 		return nil, fmt.Errorf("recover malicious %w", err)
@@ -39,7 +58,7 @@
 
 	if types.GetEffectiveGenesis() != types.FirstEffectiveGenesis() {
 		// need to load the golden atxs after a checkpoint recovery
-		if err := recoverEpoch(types.GetEffectiveGenesis().Add(1).GetEpoch(), trtl, db, beacon); err != nil {
+		if err := recoverEpoch(types.GetEffectiveGenesis().Add(1).GetEpoch(), trtl, db); err != nil {
 			return nil, err
 		}
 	}
@@ -51,42 +70,65 @@
 	epoch++ // recoverEpoch expects target epoch, rather than publish
 	if last.GetEpoch() != epoch {
 		for eid := last.GetEpoch(); eid <= epoch; eid++ {
-			if err := recoverEpoch(eid, trtl, db, beacon); err != nil {
+			if err := recoverEpoch(eid, trtl, db); err != nil {
 				return nil, err
 			}
 		}
 	}
-	start := types.GetEffectiveGenesis().Add(1)
 	for lid := start; !lid.After(last); lid = lid.Add(1) {
 		select {
 		case <-ctx.Done():
 			return nil, ctx.Err()
 		default:
 		}
-		if err := RecoverLayer(ctx, trtl, db, beacon, start, lid, last, min(last, latest)); err != nil {
+		if err := RecoverLayer(ctx, trtl, db, lid, trtl.OnRecoveredBallot); err != nil {
 			return nil, fmt.Errorf("failed to load tortoise state at layer %d: %w", lid, err)
 		}
 	}
+	if last == 0 {
+		last = current
+	} else {
+		last = min(last, current)
+	}
+	if last < start {
+		return trtl, nil
+	}
+	trtl.TallyVotes(ctx, last)
+	// find topmost layer that was already applied and reset pending
+	// so that result for that layer is not returned
+	for prev := last - 1; prev >= start; prev-- {
+		opinion, err := layers.GetAggregatedHash(db, prev)
+		if err == nil && opinion != types.EmptyLayerHash {
+			if trtl.resetPending(prev, opinion) {
+				break
+			}
+		}
+		if err != nil && !errors.Is(err, sql.ErrNotFound) {
+			return nil, fmt.Errorf("check opinion %w", err)
+		}
+	}
 	return trtl, nil
 }
 
-func recoverEpoch(epoch types.EpochID, trtl *Tortoise, db *datastore.CachedDB, beacondb system.BeaconGetter) error {
+func recoverEpoch(epoch types.EpochID, trtl *Tortoise, db *datastore.CachedDB) error {
 	if err := db.IterateEpochATXHeaders(epoch, func(header *types.ActivationTxHeader) error {
 		trtl.OnAtx(header.ToData())
 		return nil
 	}); err != nil {
 		return err
 	}
-	beacon, err := beacondb.GetBeacon(epoch)
-	if err == nil {
+	beacon, err := beacons.Get(db, epoch)
+	if err == nil && beacon != types.EmptyBeacon {
 		trtl.OnBeacon(epoch, beacon)
 	}
 	return nil
 }
 
-func RecoverLayer(ctx context.Context, trtl *Tortoise, db *datastore.CachedDB, beacon system.BeaconGetter, start, lid, last, current types.LayerID) error {
+type ballotFunc func(*types.BallotTortoiseData)
+
+func RecoverLayer(ctx context.Context, trtl *Tortoise, db *datastore.CachedDB, lid types.LayerID, onBallot ballotFunc) error {
 	if lid.FirstInEpoch() {
-		if err := recoverEpoch(lid.GetEpoch(), trtl, db, beacon); err != nil {
+		if err := recoverEpoch(lid.GetEpoch(), trtl, db); err != nil {
 			return err
 		}
 	}
@@ -119,36 +161,19 @@
 	}
 	for _, ballot := range ballotsrst {
 		if ballot.EpochData != nil {
-			trtl.OnBallot(ballot.ToTortoiseData())
+			onBallot(ballot.ToTortoiseData())
 		}
 	}
 	for _, ballot := range ballotsrst {
 		if ballot.EpochData == nil {
-			trtl.OnBallot(ballot.ToTortoiseData())
+			onBallot(ballot.ToTortoiseData())
 		}
 	}
 	coin, err := layers.GetWeakCoin(db, lid)
 	if err != nil && !errors.Is(err, sql.ErrNotFound) {
 		return err
-	}
-	if err == nil {
+	} else if err == nil {
 		trtl.OnWeakCoin(lid, coin)
 	}
-	if lid <= current && (lid%types.LayerID(trtl.cfg.WindowSize) == 0 || lid == last) {
-		trtl.TallyVotes(ctx, lid)
-		// find topmost layer that was already applied and reset pending
-		// so that result for that layer is not returned
-		for prev := lid - 1; prev >= start; prev-- {
-			opinion, err := layers.GetAggregatedHash(db, prev)
-			if err == nil {
-				if trtl.resetPending(prev, opinion) {
-					return nil
-				}
-			} else if !errors.Is(err, sql.ErrNotFound) {
-				return fmt.Errorf("check opinion %w", err)
-			}
-		}
-
-	}
 	return nil
 }