storage: add a migration for missing HardState

See cockroachdb#6991. It's possible that the HardState is missing after a snapshot was
applied (so there is a TruncatedState). In this case, synthesize a HardState
(simply setting everything that was in the snapshot to committed). Having lost
the original HardState can theoretically mean that the replica was further
ahead or had voted, and so there's no guarantee that this will be correct. But
it will be correct in the majority of cases, and some state *has* to be
recovered.

To illustrate this in the scenario in cockroachdb#6991: There, we (presumably) have
applied an empty snapshot (no real data, but a Raft log which starts and
ends at index ten as designated by its TruncatedState). We don't have a
HardState, so Raft will crash because its Commit index zero isn't in line
with the fact that our Raft log starts only at index ten.

The migration sees that there is a TruncatedState, but no HardState. It will
synthesize a HardState with Commit:10 (and the corresponding Term from the
TruncatedState, which is five).

storage/store.go

@@ -866,38 +866,55 @@ func IterateRangeDescriptors(
 	return err
 }
 
-// MIGRATION(tschottdorf): As of #7310, we make sure that a Replica
-// always has a complete Raft state on disk. Prior versions may not
-// have that, which causes issues due to the fact that we used to
-// synthesize a TruncatedState and do so no more. To make up for
-// that, write a missing TruncatedState here. That key is in the
-// replicated state, but since during a cluster upgrade, all nodes
-// do it, it's fine (and we never CPut on that key, so anything in
-// the Raft pipeline will simply overwrite it).
+// MIGRATION(tschottdorf): As of #7310, we make sure that a Replica always has
+// a complete Raft state on disk. Prior versions may not have that, which
+// causes issues due to the fact that we used to synthesize a TruncatedState
+// and do so no more. To make up for that, write a missing TruncatedState here.
+// That key is in the replicated state, but since during a cluster upgrade, all
+// nodes do it, it's fine (and we never CPut on that key, so anything in the
+// Raft pipeline will simply overwrite it).
 //
-// TODO(tschottdorf): test this method.
-func (s *Store) migrate7310(desc roachpb.RangeDescriptor) {
+// Migration(tschottdorf): See #6991. It's possible that the HardState is
+// missing after a snapshot was applied (so there is a TruncatedState). In this
+// case, synthesize a HardState (simply setting everything that was in the
+// snapshot to committed). Having lost the original HardState can theoretically
+// mean that the replica was further ahead or had voted, and so there's no
+// guarantee that this will be correct. But it will be correct in the majority
+// of cases, and some state *has* to be recovered.
+func (s *Store) migrate7310And6991(desc roachpb.RangeDescriptor) {
 	if !desc.IsInitialized() {
 		log.Fatalf("found uninitialized descriptor on range: %+v", desc)
 	}
 	batch := s.engine.NewBatch()
 	state, err := loadState(batch, &desc)
 	if err != nil {
-		log.Fatalf("could not migrate truncated state: %s", err)
+		log.Fatal(errors.Wrap(err, "could not migrate TruncatedState: %s"))
 	}
-	if (*state.TruncatedState != roachpb.RaftTruncatedState{}) {
-		return
+	if (*state.TruncatedState == roachpb.RaftTruncatedState{}) {
+		state.TruncatedState.Term = raftInitialLogTerm
+		state.TruncatedState.Index = raftInitialLogIndex
+		if _, err := saveState(batch, state); err != nil {
+			log.Fatal(errors.Wrapf(err, "could not migrate TruncatedState to %+v", &state.TruncatedState))
+		}
+		log.Warningf("migration: synthesized TruncatedState for %+v", desc)
 	}
 
-	state.TruncatedState.Term = raftInitialLogTerm
-	state.TruncatedState.Index = raftInitialLogIndex
-	if _, err := saveState(batch, state); err != nil {
-		log.Fatalf("could not migrate truncated state: %s", err)
+	hs, err := loadHardState(batch, desc.RangeID)
+	if err != nil {
+		log.Fatal(errors.Wrap(err, "unable to load HardState"))
+	}
+	// Only update the HardState when there is a nontrivial Commit field. We
+	// don't have a snapshot here, so we could wind up lowering the commit
+	// index (which would error out and fatal us).
+	if hs.Commit == 0 {
+		if err := synthesizeHardState(batch, state); err != nil {
+			log.Fatal(errors.Wrap(err, "could not migrate HardState"))
+		}
 	}
+
 	if err := batch.Commit(); err != nil {
-		log.Fatalf("could not migrate truncated state: %s", err)
+		log.Fatalf("could not migrate Raft state: %s", err)
 	}
-	log.Warningf("migration: synthesized truncated state for %+v", desc)
 }
 
 // Start the engine, set the GC and read the StoreIdent.
@@ -987,7 +1004,7 @@ func (s *Store) Start(stopper *stop.Stopper) error {
 			return false, s.destroyReplicaData(&desc)
 		}
 
-		s.migrate7310(desc)
+		s.migrate7310And6991(desc)
 
 		rng, err := NewReplica(&desc, s, 0)
 		if err != nil {