Safe delete states (#4401)

* Filter block roots by finalization and head
* Tests
* Comments
* Merge branch 'master' into safe-delete-states
* Fixed exisiting tests
* Merge branch 'safe-delete-states' of git+ssh://github.com/prysmaticlabs/prysm into safe-delete-states
* Merge refs/heads/master into safe-delete-states
* Merge refs/heads/master into safe-delete-states
* Merge refs/heads/master into safe-delete-states
* Merge refs/heads/master into safe-delete-states

beacon-chain/blockchain/forkchoice/process_block.go

@@ -468,6 +468,11 @@ func (s *Store) rmStatesOlderThanLastFinalized(ctx context.Context, startSlot ui
 		return err
 	}
 
+	roots, err = s.filterBlockRoots(ctx, roots)
+	if err != nil {
+		return err
+	}
+
 	if err := s.db.DeleteStates(ctx, roots); err != nil {
 		return err
 	}
@@ -522,3 +527,31 @@ func (s *Store) saveInitState(ctx context.Context, state *pb.BeaconState) error
 	}
 	return nil
 }
+
+// This filters block roots that are not known as head root and finalized root in DB.
+// It serves as the last line of defence before we prune states.
+func (s *Store) filterBlockRoots(ctx context.Context, roots [][32]byte) ([][32]byte, error) {
+	f, err := s.db.FinalizedCheckpoint(ctx)
+	if err != nil {
+		return nil, err
+	}
+	fRoot := f.Root
+	h, err := s.db.HeadBlock(ctx)
+	if err != nil {
+		return nil, err
+	}
+	hRoot, err := ssz.SigningRoot(h)
+	if err != nil {
+		return nil, err
+	}
+
+	filtered := make([][32]byte, 0, len(roots))
+	for _, root := range roots {
+		if bytes.Equal(root[:], fRoot[:]) || bytes.Equal(root[:], hRoot[:]) {
+			continue
+		}
+		filtered = append(filtered, root)
+	}
+
+	return filtered, nil
+}

beacon-chain/blockchain/forkchoice/process_block_test.go

@@ -289,6 +289,9 @@ func TestRemoveStateSinceLastFinalized(t *testing.T) {
 			t.Fatal(err)
 		}
 		blockRoots = append(blockRoots, r)
+		if err := store.db.SaveHeadBlockRoot(ctx, r); err != nil {
+			t.Fatal(err)
+		}
 	}
 
 	// New finalized epoch: 1
@@ -361,6 +364,9 @@ func TestRemoveStateSinceLastFinalized_EmptyStartSlot(t *testing.T) {
 		}
 		blockRoots = append(blockRoots, r)
 	}
+	if err := store.db.SaveHeadBlockRoot(ctx, blockRoots[0]); err != nil {
+		t.Fatal(err)
+	}
 	if err := store.rmStatesOlderThanLastFinalized(ctx, 10, 11); err != nil {
 		t.Fatal(err)
 	}
@@ -483,3 +489,45 @@ func TestSaveInitState_CanSaveDelete(t *testing.T) {
 		t.Errorf("wanted: %d, got: %d", len(store.initSyncState), params.BeaconConfig().SlotsPerEpoch)
 	}
 }
+
+func TestFilterBlockRoots_CanFilter(t *testing.T) {
+	ctx := context.Background()
+	db := testDB.SetupDB(t)
+	defer testDB.TeardownDB(t, db)
+
+	store := NewForkChoiceService(ctx, db)
+	fBlock := &ethpb.BeaconBlock{}
+	fRoot, _ := ssz.SigningRoot(fBlock)
+	hBlock := &ethpb.BeaconBlock{Slot: 1}
+	headRoot, _ := ssz.SigningRoot(hBlock)
+	if err := store.db.SaveBlock(ctx, fBlock); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.db.SaveState(ctx, &pb.BeaconState{}, fRoot); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.db.SaveFinalizedCheckpoint(ctx, &ethpb.Checkpoint{Root: fRoot[:]}); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.db.SaveBlock(ctx, hBlock); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.db.SaveState(ctx, &pb.BeaconState{}, headRoot); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.db.SaveHeadBlockRoot(ctx, headRoot); err != nil {
+		t.Fatal(err)
+	}
+
+	roots := [][32]byte{{'C'}, {'D'}, headRoot, {'E'}, fRoot, {'F'}}
+	wanted := [][32]byte{{'C'}, {'D'}, {'E'}, {'F'}}
+
+	received, err := store.filterBlockRoots(ctx, roots)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(wanted, received) {
+		t.Error("Did not filter correctly")
+	}
+}