diff --git a/pkg/storage/raft.pb.go b/pkg/storage/raft.pb.go
index 97b93abcba9a..05c6e6ce9c58 100644
--- a/pkg/storage/raft.pb.go
+++ b/pkg/storage/raft.pb.go
@@ -70,7 +70,7 @@ func (x *SnapshotRequest_Priority) UnmarshalJSON(data []byte) error {
 	return nil
 }
 func (SnapshotRequest_Priority) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{5, 0}
+	return fileDescriptor_raft_c419318fe988e310, []int{5, 0}
 }
 
 type SnapshotRequest_Strategy int32
@@ -107,7 +107,7 @@ func (x *SnapshotRequest_Strategy) UnmarshalJSON(data []byte) error {
 	return nil
 }
 func (SnapshotRequest_Strategy) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{5, 1}
+	return fileDescriptor_raft_c419318fe988e310, []int{5, 1}
 }
 
 type SnapshotRequest_Type int32
@@ -146,7 +146,7 @@ func (x *SnapshotRequest_Type) UnmarshalJSON(data []byte) error {
 	return nil
 }
 func (SnapshotRequest_Type) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{5, 2}
+	return fileDescriptor_raft_c419318fe988e310, []int{5, 2}
 }
 
 type SnapshotResponse_Status int32
@@ -191,7 +191,7 @@ func (x *SnapshotResponse_Status) UnmarshalJSON(data []byte) error {
 	return nil
 }
 func (SnapshotResponse_Status) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{6, 0}
+	return fileDescriptor_raft_c419318fe988e310, []int{6, 0}
 }
 
 // RaftHeartbeat is a request that contains the barebones information for a
@@ -205,13 +205,22 @@ type RaftHeartbeat struct {
 	Term          uint64                                                 `protobuf:"varint,4,opt,name=term" json:"term"`
 	Commit        uint64                                                 `protobuf:"varint,5,opt,name=commit" json:"commit"`
 	Quiesce       bool                                                   `protobuf:"varint,6,opt,name=quiesce" json:"quiesce"`
+	// ToIsLearner was added in v19.2 to aid in the transition from preemptive
+	// snapshots to learner replicas. If a Replica learns its ID from a message
+	// which indicates that it is a learner and it is not currently a part of the
+	// range (due to being from a preemptive snapshot) then it must delete all of
+	// its data.
+	//
+	// TODO(ajwerner): remove in 20.2 once we ensure that preemptive snapshots can
+	// no longer be present and that we're never talking to a 19.2 node.
+	ToIsLearner bool `protobuf:"varint,7,opt,name=to_is_learner,json=toIsLearner" json:"to_is_learner"`
 }
 
 func (m *RaftHeartbeat) Reset()         { *m = RaftHeartbeat{} }
 func (m *RaftHeartbeat) String() string { return proto.CompactTextString(m) }
 func (*RaftHeartbeat) ProtoMessage()    {}
 func (*RaftHeartbeat) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{0}
+	return fileDescriptor_raft_c419318fe988e310, []int{0}
 }
 func (m *RaftHeartbeat) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -267,7 +276,7 @@ func (m *RaftMessageRequest) Reset()         { *m = RaftMessageRequest{} }
 func (m *RaftMessageRequest) String() string { return proto.CompactTextString(m) }
 func (*RaftMessageRequest) ProtoMessage()    {}
 func (*RaftMessageRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{1}
+	return fileDescriptor_raft_c419318fe988e310, []int{1}
 }
 func (m *RaftMessageRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -300,7 +309,7 @@ func (m *RaftMessageRequestBatch) Reset()         { *m = RaftMessageRequestBatch
 func (m *RaftMessageRequestBatch) String() string { return proto.CompactTextString(m) }
 func (*RaftMessageRequestBatch) ProtoMessage()    {}
 func (*RaftMessageRequestBatch) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{2}
+	return fileDescriptor_raft_c419318fe988e310, []int{2}
 }
 func (m *RaftMessageRequestBatch) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -333,7 +342,7 @@ func (m *RaftMessageResponseUnion) Reset()         { *m = RaftMessageResponseUni
 func (m *RaftMessageResponseUnion) String() string { return proto.CompactTextString(m) }
 func (*RaftMessageResponseUnion) ProtoMessage()    {}
 func (*RaftMessageResponseUnion) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{3}
+	return fileDescriptor_raft_c419318fe988e310, []int{3}
 }
 func (m *RaftMessageResponseUnion) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -376,7 +385,7 @@ func (m *RaftMessageResponse) Reset()         { *m = RaftMessageResponse{} }
 func (m *RaftMessageResponse) String() string { return proto.CompactTextString(m) }
 func (*RaftMessageResponse) ProtoMessage()    {}
 func (*RaftMessageResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{4}
+	return fileDescriptor_raft_c419318fe988e310, []int{4}
 }
 func (m *RaftMessageResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -417,7 +426,7 @@ func (m *SnapshotRequest) Reset()         { *m = SnapshotRequest{} }
 func (m *SnapshotRequest) String() string { return proto.CompactTextString(m) }
 func (*SnapshotRequest) ProtoMessage()    {}
 func (*SnapshotRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{5}
+	return fileDescriptor_raft_c419318fe988e310, []int{5}
 }
 func (m *SnapshotRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -479,7 +488,7 @@ func (m *SnapshotRequest_Header) Reset()         { *m = SnapshotRequest_Header{}
 func (m *SnapshotRequest_Header) String() string { return proto.CompactTextString(m) }
 func (*SnapshotRequest_Header) ProtoMessage()    {}
 func (*SnapshotRequest_Header) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{5, 0}
+	return fileDescriptor_raft_c419318fe988e310, []int{5, 0}
 }
 func (m *SnapshotRequest_Header) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -513,7 +522,7 @@ func (m *SnapshotResponse) Reset()         { *m = SnapshotResponse{} }
 func (m *SnapshotResponse) String() string { return proto.CompactTextString(m) }
 func (*SnapshotResponse) ProtoMessage()    {}
 func (*SnapshotResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{6}
+	return fileDescriptor_raft_c419318fe988e310, []int{6}
 }
 func (m *SnapshotResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -550,7 +559,7 @@ func (m *ConfChangeContext) Reset()         { *m = ConfChangeContext{} }
 func (m *ConfChangeContext) String() string { return proto.CompactTextString(m) }
 func (*ConfChangeContext) ProtoMessage()    {}
 func (*ConfChangeContext) Descriptor() ([]byte, []int) {
-	return fileDescriptor_raft_d6a64da2f1251cf6, []int{7}
+	return fileDescriptor_raft_c419318fe988e310, []int{7}
 }
 func (m *ConfChangeContext) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -798,6 +807,14 @@ func (m *RaftHeartbeat) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0
 	}
 	i++
+	dAtA[i] = 0x38
+	i++
+	if m.ToIsLearner {
+		dAtA[i] = 1
+	} else {
+		dAtA[i] = 0
+	}
+	i++
 	return i, nil
 }
 
@@ -1173,6 +1190,7 @@ func (m *RaftHeartbeat) Size() (n int) {
 	n += 1 + sovRaft(uint64(m.Term))
 	n += 1 + sovRaft(uint64(m.Commit))
 	n += 2
+	n += 2
 	return n
 }
 
@@ -1496,6 +1514,26 @@ func (m *RaftHeartbeat) Unmarshal(dAtA []byte) error {
 				}
 			}
 			m.Quiesce = bool(v != 0)
+		case 7:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ToIsLearner", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowRaft
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.ToIsLearner = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := skipRaft(dAtA[iNdEx:])
@@ -2814,85 +2852,86 @@ var (
 	ErrIntOverflowRaft   = fmt.Errorf("proto: integer overflow")
 )
 
-func init() { proto.RegisterFile("storage/raft.proto", fileDescriptor_raft_d6a64da2f1251cf6) }
-
-var fileDescriptor_raft_d6a64da2f1251cf6 = []byte{
-	// 1224 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x56, 0x5f, 0x6f, 0xdb, 0x54,
-	0x14, 0x8f, 0x1b, 0x27, 0x71, 0x4e, 0x92, 0xd6, 0xbb, 0x4c, 0x60, 0x85, 0x91, 0x46, 0x1e, 0x1b,
-	0xd9, 0x10, 0xc9, 0xa8, 0x06, 0x0f, 0xbc, 0xe5, 0x8f, 0xbb, 0x66, 0xfd, 0x17, 0xb9, 0x59, 0x11,
-	0x48, 0x53, 0xe4, 0x38, 0xb7, 0x89, 0xd5, 0xc4, 0xd7, 0xb3, 0x6f, 0x06, 0xd9, 0xa7, 0xe0, 0x23,
-	0xf0, 0xca, 0x27, 0xe0, 0x2b, 0xf4, 0x05, 0x69, 0x8f, 0x93, 0x40, 0x15, 0x74, 0x7c, 0x8a, 0x3d,
-	0xa1, 0xfb, 0xc7, 0x69, 0xd6, 0x06, 0xd6, 0x4a, 0x88, 0x17, 0x5e, 0x12, 0xfb, 0x9c, 0xf3, 0xfb,
-	0x1d, 0x9f, 0xbf, 0xf7, 0x02, 0x8a, 0x28, 0x09, 0x9d, 0x21, 0xae, 0x85, 0xce, 0x11, 0xad, 0x06,
-	0x21, 0xa1, 0x04, 0xdd, 0x70, 0x89, 0x7b, 0x1c, 0x12, 0xc7, 0x1d, 0x55, 0xa5, 0xb6, 0x78, 0x93,
-	0xbf, 0x06, 0xfd, 0x1a, 0x0e, 0x43, 0x12, 0x46, 0xc2, 0xb0, 0xf8, 0x7e, 0x2c, 0x9d, 0x60, 0xea,
-	0x0c, 0x1c, 0xea, 0x48, 0xf9, 0x47, 0x31, 0xa9, 0xfc, 0x0f, 0xfa, 0xb5, 0x88, 0x3a, 0x14, 0x4b,
-	0xf5, 0x87, 0x98, 0xba, 0x03, 0xee, 0x90, 0xff, 0x04, 0xfd, 0x05, 0xe7, 0xc5, 0x9b, 0x43, 0x32,
-	0x24, 0xfc, 0xb1, 0xc6, 0x9e, 0x84, 0xd4, 0xfc, 0x39, 0x09, 0x05, 0xdb, 0x39, 0xa2, 0x5b, 0xd8,
-	0x09, 0x69, 0x1f, 0x3b, 0x14, 0xf5, 0x41, 0x0b, 0x1d, 0x7f, 0x88, 0x7b, 0xde, 0xc0, 0x50, 0xca,
-	0x4a, 0x45, 0x6d, 0x3c, 0x3a, 0x39, 0x5d, 0x4f, 0x9c, 0x9d, 0xae, 0x67, 0x6c, 0x26, 0x6f, 0xb7,
-	0xde, 0x9c, 0xae, 0x3f, 0x1c, 0x7a, 0x74, 0x34, 0xed, 0x57, 0x5d, 0x32, 0xa9, 0xcd, 0x83, 0x1a,
-	0xf4, 0xcf, 0x9f, 0x6b, 0xc1, 0xf1, 0xb0, 0x26, 0xa3, 0xa8, 0x4a, 0x9c, 0x9d, 0xe1, 0xc4, 0xed,
-	0x01, 0xfa, 0x0e, 0xd6, 0x8e, 0x42, 0x32, 0xe9, 0x85, 0x38, 0x18, 0x7b, 0xae, 0xc3, 0x5c, 0xad,
-	0x94, 0x95, 0x4a, 0xa1, 0xb1, 0x2f, 0x5d, 0x15, 0x36, 0x43, 0x32, 0xb1, 0x85, 0x96, 0x3b, 0xfc,
-	0xf2, 0x7a, 0x0e, 0x63, 0xa4, 0x5d, 0x38, 0x5a, 0x20, 0x1a, 0xa0, 0x67, 0x50, 0xa0, 0x64, 0xd1,
-	0x6d, 0x92, 0xbb, 0xdd, 0x95, 0x6e, 0x73, 0x5d, 0xf2, 0x6f, 0x38, 0xcd, 0x51, 0x72, 0xee, 0xd2,
-	0x00, 0x95, 0xe2, 0x70, 0x62, 0xa8, 0x3c, 0x97, 0x2a, 0xf3, 0x64, 0x73, 0x09, 0xba, 0x05, 0x69,
-	0x97, 0x4c, 0x26, 0x1e, 0x35, 0x52, 0x0b, 0x3a, 0x29, 0x43, 0x25, 0xc8, 0x3c, 0x9b, 0x7a, 0x38,
-	0x72, 0xb1, 0x91, 0x2e, 0x2b, 0x15, 0x4d, 0xaa, 0x63, 0xa1, 0xf9, 0xab, 0x0a, 0x88, 0x55, 0x6e,
-	0x17, 0x47, 0x91, 0x33, 0xc4, 0x36, 0x7e, 0x36, 0xc5, 0xd1, 0x7f, 0x53, 0xbe, 0x5d, 0xc8, 0x2f,
-	0x96, 0x8f, 0xd7, 0x2e, 0xb7, 0xf1, 0x71, 0xf5, 0xbc, 0xbd, 0x2f, 0xe4, 0xa4, 0x85, 0x23, 0x37,
-	0xf4, 0x02, 0x4a, 0x42, 0x19, 0x45, 0x6e, 0xa1, 0x2c, 0xa8, 0x0d, 0x70, 0x5e, 0x14, 0x5e, 0x91,
-	0xeb, 0x91, 0x65, 0xe7, 0xe9, 0x46, 0x35, 0xc8, 0x4c, 0x44, 0x3e, 0x78, 0xbe, 0x73, 0x1b, 0x6b,
-	0x55, 0x31, 0x09, 0x55, 0x99, 0xa6, 0x38, 0x8b, 0xd2, 0x6a, 0x31, 0xcb, 0xa9, 0x25, 0x59, 0x46,
-	0x9b, 0x00, 0xa3, 0x78, 0x34, 0x22, 0x23, 0x5d, 0x4e, 0x56, 0x72, 0x1b, 0xe5, 0xea, 0xa5, 0x39,
-	0xae, 0xbe, 0x35, 0x43, 0x92, 0x64, 0x01, 0x89, 0xf6, 0x61, 0x6d, 0xfe, 0xd6, 0x0b, 0x71, 0x14,
-	0x44, 0x46, 0xe6, 0x5a, 0x64, 0xab, 0x73, 0xb8, 0xcd, 0xd0, 0xe8, 0x29, 0xac, 0x89, 0x3a, 0x47,
-	0xd4, 0x09, 0x69, 0xef, 0x18, 0xcf, 0x0c, 0xad, 0xac, 0x54, 0xf2, 0x8d, 0x2f, 0xde, 0x9c, 0xae,
-	0x7f, 0x7e, 0xbd, 0xfa, 0x6e, 0xe3, 0x99, 0x5d, 0xe0, 0x6c, 0x07, 0x8c, 0x6c, 0x1b, 0xcf, 0xcc,
-	0x3e, 0x7c, 0x70, 0xb9, 0xb9, 0x1a, 0x0e, 0x75, 0x47, 0xe8, 0x11, 0x68, 0xa1, 0x78, 0x8f, 0x0c,
-	0x85, 0xc7, 0x70, 0xe7, 0x6f, 0x62, 0xb8, 0x80, 0x16, 0x81, 0xcc, 0xc1, 0x66, 0x07, 0x8c, 0xb7,
-	0xac, 0xa2, 0x80, 0xf8, 0x11, 0x7e, 0xe2, 0x7b, 0xc4, 0x47, 0x55, 0x48, 0xf1, 0x8d, 0xc8, 0x7b,
-	0x38, 0xb7, 0x61, 0x2c, 0x69, 0x07, 0x8b, 0xe9, 0x6d, 0x61, 0xf6, 0x95, 0x7a, 0xf2, 0xe3, 0xba,
-	0x62, 0xfe, 0xb6, 0x02, 0xef, 0x2d, 0xa1, 0xfc, 0x9f, 0x0f, 0xc5, 0x23, 0x48, 0x4d, 0x59, 0x52,
-	0xe5, 0x48, 0x7c, 0xfa, 0xae, 0x6a, 0x2d, 0xd4, 0x41, 0x92, 0x09, 0xbc, 0xf9, 0x67, 0x1a, 0xd6,
-	0x0e, 0x7c, 0x27, 0x88, 0x46, 0x84, 0xc6, 0xfb, 0xa6, 0x0e, 0xe9, 0x11, 0x76, 0x06, 0x38, 0xae,
-	0xd4, 0xbd, 0x25, 0xec, 0x17, 0x30, 0xd5, 0x2d, 0x0e, 0xb0, 0x25, 0x10, 0xdd, 0x05, 0xed, 0xf8,
-	0x79, 0xaf, 0xcf, 0x9a, 0x8b, 0x67, 0x2d, 0xdf, 0xc8, 0xb1, 0xca, 0x6c, 0x1f, 0xf2, 0x7e, 0xb3,
-	0x33, 0xc7, 0xcf, 0x45, 0xe3, 0xad, 0x43, 0x6e, 0x4c, 0x86, 0x3d, 0xec, 0xd3, 0xd0, 0xc3, 0x91,
-	0x91, 0x2c, 0x27, 0x2b, 0x79, 0x1b, 0xc6, 0x64, 0x68, 0x09, 0x09, 0x2a, 0x42, 0xea, 0xc8, 0xf3,
-	0x9d, 0x31, 0x0f, 0x34, 0x1e, 0x65, 0x21, 0x2a, 0xfe, 0xa4, 0x42, 0x5a, 0xf8, 0x45, 0x4f, 0xe1,
-	0x26, 0x5b, 0x0a, 0x3d, 0xb9, 0x03, 0x7a, 0xb2, 0x21, 0x65, 0xc5, 0xae, 0xd5, 0xcc, 0x28, 0xbc,
-	0xbc, 0x81, 0x6f, 0x03, 0xc8, 0xc9, 0xf4, 0x5e, 0x60, 0x5e, 0xb9, 0x64, 0x5c, 0x13, 0x31, 0x63,
-	0xde, 0x0b, 0x8c, 0xee, 0x40, 0xce, 0x75, 0xfc, 0xde, 0x00, 0xbb, 0x63, 0xcf, 0xc7, 0x6f, 0x7d,
-	0x30, 0xb8, 0x8e, 0xdf, 0x12, 0x72, 0x64, 0x41, 0x8a, 0x1f, 0xf0, 0x7c, 0x39, 0x2d, 0x4f, 0xee,
-	0xfc, 0x2a, 0x10, 0xb7, 0xc2, 0x01, 0x03, 0xc4, 0xc1, 0x73, 0x34, 0xda, 0x05, 0x2d, 0x08, 0x3d,
-	0x12, 0x7a, 0x74, 0xc6, 0x0f, 0x93, 0xd5, 0xa5, 0x4d, 0x70, 0xb1, 0x4c, 0x1d, 0x09, 0x89, 0x07,
-	0x37, 0xa6, 0x60, 0x74, 0x11, 0x0d, 0x1d, 0x8a, 0x87, 0x33, 0x23, 0x73, 0x65, 0xba, 0x03, 0x09,
-	0x89, 0xe9, 0x62, 0x0a, 0xb4, 0x09, 0xb7, 0xa6, 0xbe, 0xec, 0x74, 0x8a, 0x07, 0x3d, 0x1a, 0x4e,
-	0x7d, 0xf1, 0x24, 0x62, 0xd7, 0x16, 0x92, 0x53, 0x5c, 0xb4, 0xec, 0xc6, 0x86, 0x3c, 0x64, 0x54,
-	0x07, 0x95, 0xce, 0x02, 0x6c, 0x64, 0xf9, 0x27, 0x7d, 0x72, 0x85, 0x4f, 0xea, 0xce, 0x02, 0x3c,
-	0x3f, 0x92, 0x67, 0x01, 0x7e, 0xac, 0x6a, 0x8a, 0xbe, 0x62, 0x3e, 0x04, 0x2d, 0x8e, 0x1d, 0xe5,
-	0x20, 0xf3, 0x64, 0x6f, 0x7b, 0x6f, 0xff, 0xeb, 0x3d, 0x3d, 0x81, 0xf2, 0xa0, 0xd9, 0x56, 0x73,
-	0xff, 0xd0, 0xb2, 0xbf, 0xd1, 0x15, 0x54, 0x80, 0xac, 0x6d, 0x35, 0xea, 0x3b, 0xf5, 0xbd, 0xa6,
-	0xa5, 0xaf, 0x98, 0x06, 0x68, 0x71, 0x88, 0xcc, 0x70, 0xfb, 0xb0, 0xd7, 0xa8, 0x77, 0x9b, 0x5b,
-	0x7a, 0xc2, 0xfc, 0x0c, 0x54, 0xe6, 0x09, 0x69, 0xa0, 0xda, 0xf5, 0xcd, 0xae, 0x9e, 0x60, 0xac,
-	0x3b, 0x56, 0xdd, 0xde, 0xb3, 0x6c, 0x5d, 0x41, 0xab, 0x00, 0x1d, 0xdb, 0xb2, 0x76, 0x3b, 0xdd,
-	0xf6, 0x21, 0x23, 0xfa, 0x45, 0x01, 0xfd, 0xfc, 0x4b, 0xe5, 0x0a, 0xdb, 0x82, 0x34, 0xcb, 0xc6,
-	0x34, 0xe2, 0x73, 0xb6, 0xba, 0x71, 0xff, 0x1f, 0xc3, 0x13, 0xa0, 0xea, 0x01, 0x47, 0xc4, 0x17,
-	0x0b, 0x81, 0x67, 0x47, 0x5e, 0x7c, 0x46, 0xb2, 0x8e, 0xcf, 0x5e, 0x38, 0x12, 0xcd, 0x36, 0xa4,
-	0x05, 0xee, 0x52, 0xec, 0xf5, 0x66, 0xd3, 0xea, 0x74, 0xad, 0x96, 0xae, 0x30, 0x55, 0xbd, 0xd3,
-	0xd9, 0x69, 0x5b, 0x2d, 0x7d, 0x05, 0x65, 0x21, 0x65, 0xd9, 0xf6, 0xbe, 0xad, 0x27, 0x99, 0x55,
-	0xcb, 0x6a, 0xee, 0xb4, 0xf7, 0xac, 0x96, 0xae, 0x3e, 0x56, 0xb5, 0xa4, 0xae, 0x9a, 0x3d, 0xb8,
-	0xd1, 0x24, 0xfe, 0x51, 0x73, 0xc4, 0xba, 0xbf, 0x49, 0x7c, 0x8a, 0xbf, 0xa7, 0xe8, 0x01, 0x00,
-	0xbb, 0xe8, 0x38, 0xfe, 0x20, 0x5e, 0xca, 0xd9, 0xc6, 0x0d, 0xb9, 0x94, 0xb3, 0x4d, 0xa1, 0x69,
-	0xb7, 0xec, 0xac, 0x34, 0xe2, 0x17, 0xa9, 0x4c, 0xe0, 0xcc, 0xc6, 0xc4, 0x11, 0x97, 0xc5, 0xbc,
-	0x1d, 0xbf, 0x6e, 0xbc, 0x52, 0x20, 0xbb, 0x3b, 0x1d, 0x53, 0x8f, 0xcd, 0x29, 0x1a, 0x83, 0xbe,
-	0x30, 0xaf, 0x62, 0x75, 0xdc, 0xbf, 0xda, 0x50, 0x33, 0xdb, 0xe2, 0xdd, 0xab, 0xed, 0x47, 0x33,
-	0x51, 0x51, 0x1e, 0x28, 0xe8, 0x29, 0xe4, 0x99, 0x32, 0x4e, 0x3d, 0x32, 0xdf, 0xdd, 0x76, 0xc5,
-	0xdb, 0x57, 0xa8, 0x9d, 0xa0, 0x6f, 0xdc, 0x3b, 0xf9, 0xa3, 0x94, 0x38, 0x39, 0x2b, 0x29, 0x2f,
-	0xcf, 0x4a, 0xca, 0xab, 0xb3, 0x92, 0xf2, 0xfb, 0x59, 0x49, 0xf9, 0xe1, 0x75, 0x29, 0xf1, 0xf2,
-	0x75, 0x29, 0xf1, 0xea, 0x75, 0x29, 0xf1, 0x6d, 0x46, 0x32, 0xfc, 0x15, 0x00, 0x00, 0xff, 0xff,
-	0x38, 0x31, 0xfb, 0x9d, 0x72, 0x0c, 0x00, 0x00,
+func init() { proto.RegisterFile("storage/raft.proto", fileDescriptor_raft_c419318fe988e310) }
+
+var fileDescriptor_raft_c419318fe988e310 = []byte{
+	// 1243 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x56, 0x4b, 0x6f, 0xdb, 0x46,
+	0x10, 0x16, 0x2d, 0x5a, 0x8f, 0x91, 0x64, 0x33, 0xdb, 0xa0, 0x25, 0xd4, 0x54, 0x16, 0x94, 0x26,
+	0x55, 0x52, 0x54, 0x4a, 0x8d, 0xb4, 0x87, 0xde, 0xf4, 0xa0, 0x63, 0xc5, 0x2f, 0x81, 0x56, 0x5c,
+	0xb4, 0x40, 0x20, 0x50, 0xd4, 0x5a, 0x22, 0x2c, 0x71, 0x99, 0xe5, 0x2a, 0xad, 0xf2, 0x2b, 0xfa,
+	0x13, 0x7a, 0xed, 0x3f, 0xf1, 0xa5, 0x40, 0x8e, 0x01, 0x5a, 0x18, 0x8d, 0xd3, 0x5f, 0x91, 0x53,
+	0xb1, 0x0f, 0xca, 0x8c, 0xed, 0x36, 0x36, 0x50, 0xf4, 0xd2, 0x8b, 0x44, 0xce, 0xe3, 0x1b, 0xce,
+	0x7c, 0x33, 0xb3, 0x0b, 0x28, 0x64, 0x84, 0x3a, 0x23, 0x5c, 0xa7, 0xce, 0x21, 0xab, 0x05, 0x94,
+	0x30, 0x82, 0x6e, 0xb8, 0xc4, 0x3d, 0xa2, 0xc4, 0x71, 0xc7, 0x35, 0xa5, 0x2d, 0xde, 0x14, 0xaf,
+	0xc1, 0xa0, 0x8e, 0x29, 0x25, 0x34, 0x94, 0x86, 0xc5, 0x0f, 0x23, 0xe9, 0x14, 0x33, 0x67, 0xe8,
+	0x30, 0x47, 0xc9, 0x3f, 0x89, 0x40, 0xd5, 0x7f, 0x30, 0xa8, 0x87, 0xcc, 0x61, 0x58, 0xa9, 0x3f,
+	0xc6, 0xcc, 0x1d, 0x8a, 0x80, 0xe2, 0x27, 0x18, 0xc4, 0x82, 0x17, 0x6f, 0x8e, 0xc8, 0x88, 0x88,
+	0xc7, 0x3a, 0x7f, 0x92, 0xd2, 0xca, 0xeb, 0x24, 0x14, 0x6c, 0xe7, 0x90, 0x6d, 0x62, 0x87, 0xb2,
+	0x01, 0x76, 0x18, 0x1a, 0x40, 0x86, 0x3a, 0xfe, 0x08, 0xf7, 0xbd, 0xa1, 0xa9, 0x95, 0xb5, 0xaa,
+	0xde, 0x7c, 0x74, 0x7c, 0xb2, 0x96, 0x38, 0x3d, 0x59, 0x4b, 0xdb, 0x5c, 0xde, 0x69, 0xbf, 0x3d,
+	0x59, 0x7b, 0x38, 0xf2, 0xd8, 0x78, 0x36, 0xa8, 0xb9, 0x64, 0x5a, 0x5f, 0x24, 0x35, 0x1c, 0x9c,
+	0x3d, 0xd7, 0x83, 0xa3, 0x51, 0x5d, 0x65, 0x51, 0x53, 0x7e, 0x76, 0x5a, 0x00, 0x77, 0x86, 0xe8,
+	0x07, 0x58, 0x3d, 0xa4, 0x64, 0xda, 0xa7, 0x38, 0x98, 0x78, 0xae, 0xc3, 0x43, 0x2d, 0x95, 0xb5,
+	0x6a, 0xa1, 0xb9, 0xa7, 0x42, 0x15, 0x36, 0x28, 0x99, 0xda, 0x52, 0x2b, 0x02, 0x7e, 0x7d, 0xbd,
+	0x80, 0x91, 0xa7, 0x5d, 0x38, 0x8c, 0x01, 0x0d, 0xd1, 0x33, 0x28, 0x30, 0x12, 0x0f, 0x9b, 0x14,
+	0x61, 0x77, 0x54, 0xd8, 0x5c, 0x8f, 0xfc, 0x1b, 0x41, 0x73, 0x8c, 0x9c, 0x85, 0x34, 0x41, 0x67,
+	0x98, 0x4e, 0x4d, 0x5d, 0xd4, 0x52, 0xe7, 0x91, 0x6c, 0x21, 0x41, 0xb7, 0x20, 0xe5, 0x92, 0xe9,
+	0xd4, 0x63, 0xe6, 0x72, 0x4c, 0xa7, 0x64, 0xa8, 0x04, 0xe9, 0x67, 0x33, 0x0f, 0x87, 0x2e, 0x36,
+	0x53, 0x65, 0xad, 0x9a, 0x51, 0xea, 0x48, 0x88, 0xaa, 0x22, 0x15, 0x2f, 0xec, 0x4f, 0xb0, 0x43,
+	0x7d, 0x4c, 0xcd, 0x74, 0xcc, 0x2a, 0xc7, 0x48, 0x27, 0xdc, 0x96, 0x8a, 0xca, 0x6f, 0x3a, 0x20,
+	0xce, 0xf1, 0x0e, 0x0e, 0x43, 0x67, 0x84, 0x6d, 0xfc, 0x6c, 0x86, 0xc3, 0xff, 0x86, 0xe8, 0x1d,
+	0xc8, 0xc7, 0x89, 0x16, 0x2c, 0xe7, 0xd6, 0x3f, 0xad, 0x9d, 0x0d, 0xc2, 0xb9, 0xea, 0xb5, 0x71,
+	0xe8, 0x52, 0x2f, 0x60, 0x84, 0x46, 0x99, 0xc4, 0x08, 0x44, 0x1d, 0x80, 0x33, 0xfa, 0x04, 0x77,
+	0xd7, 0x03, 0xcb, 0x2e, 0x88, 0x41, 0x75, 0x48, 0x4f, 0x65, 0x3d, 0x04, 0x33, 0xb9, 0xf5, 0xd5,
+	0x9a, 0x9c, 0x99, 0x9a, 0x2a, 0x53, 0x54, 0x6f, 0x65, 0x15, 0xe7, 0x63, 0xf9, 0x32, 0x3e, 0x36,
+	0x00, 0xc6, 0xd1, 0x10, 0x85, 0x66, 0xaa, 0x9c, 0xac, 0xe6, 0xd6, 0xcb, 0xb5, 0x0b, 0x13, 0x5f,
+	0x7b, 0x67, 0xda, 0x14, 0x48, 0xcc, 0x13, 0xed, 0xc1, 0xea, 0xe2, 0xad, 0x4f, 0x71, 0x18, 0x84,
+	0x66, 0xfa, 0x5a, 0x60, 0x2b, 0x0b, 0x77, 0x9b, 0x7b, 0xa3, 0xa7, 0xb0, 0x2a, 0x79, 0x0e, 0x99,
+	0x43, 0x59, 0xff, 0x08, 0xcf, 0xcd, 0x4c, 0x59, 0xab, 0xe6, 0x9b, 0x5f, 0xbd, 0x3d, 0x59, 0xfb,
+	0xf2, 0x7a, 0xfc, 0x6e, 0xe1, 0xb9, 0x5d, 0x10, 0x68, 0xfb, 0x1c, 0x6c, 0x0b, 0xcf, 0x2b, 0x03,
+	0xf8, 0xe8, 0x62, 0x73, 0x35, 0x1d, 0xe6, 0x8e, 0xd1, 0x23, 0xc8, 0x50, 0xf9, 0x1e, 0x9a, 0x9a,
+	0xc8, 0xe1, 0xce, 0xdf, 0xe4, 0x70, 0xce, 0x5b, 0x26, 0xb2, 0x70, 0xae, 0x74, 0xc1, 0x7c, 0xc7,
+	0x2a, 0x0c, 0x88, 0x1f, 0xe2, 0x27, 0xbe, 0x47, 0x7c, 0x54, 0x83, 0x65, 0xb1, 0x3b, 0x45, 0x0f,
+	0xe7, 0xd6, 0xcd, 0x4b, 0xda, 0xc1, 0xe2, 0x7a, 0x5b, 0x9a, 0x7d, 0xa3, 0x1f, 0xff, 0xbc, 0xa6,
+	0x55, 0x7e, 0x5f, 0x82, 0x0f, 0x2e, 0x81, 0xfc, 0x9f, 0x0f, 0xc5, 0x23, 0x58, 0x9e, 0xf1, 0xa2,
+	0xaa, 0x91, 0xf8, 0xfc, 0x7d, 0x6c, 0xc5, 0x78, 0x50, 0x60, 0xd2, 0xbf, 0xf2, 0x67, 0x0a, 0x56,
+	0xf7, 0x7d, 0x27, 0x08, 0xc7, 0x84, 0x45, 0xfb, 0xa6, 0x01, 0xa9, 0x31, 0x76, 0x86, 0x38, 0x62,
+	0xea, 0xde, 0x25, 0xe8, 0xe7, 0x7c, 0x6a, 0x9b, 0xc2, 0xc1, 0x56, 0x8e, 0xe8, 0x2e, 0x64, 0x8e,
+	0x9e, 0xf7, 0x07, 0xbc, 0xb9, 0x44, 0xd5, 0xf2, 0xcd, 0x1c, 0x67, 0x66, 0xeb, 0x40, 0xf4, 0x9b,
+	0x9d, 0x3e, 0x7a, 0x2e, 0x1b, 0x6f, 0x0d, 0x72, 0x13, 0x32, 0xea, 0x63, 0x9f, 0x51, 0x0f, 0x87,
+	0x66, 0xb2, 0x9c, 0xac, 0xe6, 0x6d, 0x98, 0x90, 0x91, 0x25, 0x25, 0xa8, 0x08, 0xcb, 0x87, 0x9e,
+	0xef, 0x4c, 0x44, 0xa2, 0xd1, 0x28, 0x4b, 0x51, 0xf1, 0x17, 0x1d, 0x52, 0x32, 0x2e, 0x7a, 0x0a,
+	0x37, 0xf9, 0x52, 0xe8, 0xab, 0x1d, 0xd0, 0x57, 0x0d, 0xa9, 0x18, 0xbb, 0x56, 0x33, 0x23, 0x7a,
+	0x71, 0x03, 0xdf, 0x06, 0x50, 0x93, 0xe9, 0xbd, 0xc0, 0x82, 0xb9, 0x64, 0xc4, 0x89, 0x9c, 0x31,
+	0xef, 0x05, 0x46, 0x77, 0x20, 0xe7, 0x3a, 0x7e, 0x7f, 0x88, 0xdd, 0x89, 0xe7, 0xe3, 0x77, 0x3e,
+	0x18, 0x5c, 0xc7, 0x6f, 0x4b, 0x39, 0xb2, 0x60, 0x59, 0x5c, 0x05, 0xc4, 0x72, 0xba, 0xbc, 0xb8,
+	0x8b, 0x4b, 0x43, 0xd4, 0x0a, 0xfb, 0xdc, 0x21, 0x4a, 0x5e, 0x78, 0xa3, 0x1d, 0xc8, 0x04, 0xd4,
+	0x23, 0xd4, 0x63, 0x73, 0x71, 0xec, 0xac, 0x5c, 0xda, 0x04, 0xe7, 0x69, 0xea, 0x2a, 0x97, 0x68,
+	0x70, 0x23, 0x08, 0x0e, 0x17, 0x32, 0xea, 0x30, 0x3c, 0x9a, 0x8b, 0xf3, 0xe9, 0x6a, 0x70, 0xfb,
+	0xca, 0x25, 0x82, 0x8b, 0x20, 0xd0, 0x06, 0xdc, 0x9a, 0xf9, 0xaa, 0xd3, 0x19, 0x1e, 0xf6, 0x19,
+	0x9d, 0xf9, 0xf2, 0x49, 0xe6, 0x9e, 0x89, 0x15, 0xa7, 0x18, 0xb7, 0xec, 0x45, 0x86, 0x22, 0x65,
+	0xd4, 0x00, 0x9d, 0xcd, 0x03, 0x6c, 0x66, 0xc5, 0x27, 0x7d, 0x76, 0x85, 0x4f, 0xea, 0xcd, 0x03,
+	0xbc, 0x38, 0xbc, 0xe7, 0x01, 0x7e, 0xac, 0x67, 0x34, 0x63, 0xa9, 0xf2, 0x10, 0x32, 0x51, 0xee,
+	0x28, 0x07, 0xe9, 0x27, 0xbb, 0x5b, 0xbb, 0x7b, 0xdf, 0xee, 0x1a, 0x09, 0x94, 0x87, 0x8c, 0x6d,
+	0xb5, 0xf6, 0x0e, 0x2c, 0xfb, 0x3b, 0x43, 0x43, 0x05, 0xc8, 0xda, 0x56, 0xb3, 0xb1, 0xdd, 0xd8,
+	0x6d, 0x59, 0xc6, 0x52, 0xc5, 0x84, 0x4c, 0x94, 0x22, 0x37, 0xdc, 0x3a, 0xe8, 0x37, 0x1b, 0xbd,
+	0xd6, 0xa6, 0x91, 0xa8, 0x7c, 0x01, 0x3a, 0x8f, 0x84, 0x32, 0xa0, 0xdb, 0x8d, 0x8d, 0x9e, 0x91,
+	0xe0, 0xa8, 0xdb, 0x56, 0xc3, 0xde, 0xb5, 0x6c, 0x43, 0x43, 0x2b, 0x00, 0x5d, 0xdb, 0xb2, 0x76,
+	0xba, 0xbd, 0xce, 0x01, 0x07, 0xfa, 0x55, 0x03, 0xe3, 0xec, 0x4b, 0xd5, 0x0a, 0xdb, 0x84, 0x14,
+	0xaf, 0xc6, 0x2c, 0x14, 0x73, 0xb6, 0xb2, 0x7e, 0xff, 0x1f, 0xd3, 0x93, 0x4e, 0xb5, 0x7d, 0xe1,
+	0x11, 0x5d, 0x41, 0xa4, 0x3f, 0x3f, 0xf2, 0xa2, 0x33, 0x92, 0x77, 0x7c, 0xf6, 0xdc, 0x91, 0x58,
+	0xe9, 0x40, 0x4a, 0xfa, 0x5d, 0xc8, 0xbd, 0xd1, 0x6a, 0x59, 0xdd, 0x9e, 0xd5, 0x36, 0x34, 0xae,
+	0x6a, 0x74, 0xbb, 0xdb, 0x1d, 0xab, 0x6d, 0x2c, 0xa1, 0x2c, 0x2c, 0x5b, 0xb6, 0xbd, 0x67, 0x1b,
+	0x49, 0x6e, 0xd5, 0xb6, 0x5a, 0xdb, 0x9d, 0x5d, 0xab, 0x6d, 0xe8, 0x8f, 0xf5, 0x4c, 0xd2, 0xd0,
+	0x2b, 0x7d, 0xb8, 0xd1, 0x22, 0xfe, 0x61, 0x6b, 0xcc, 0xbb, 0xbf, 0x45, 0x7c, 0x86, 0x7f, 0x64,
+	0xe8, 0x01, 0x00, 0xbf, 0x12, 0x39, 0xfe, 0x30, 0x5a, 0xca, 0xd9, 0xe6, 0x0d, 0xb5, 0x94, 0xb3,
+	0x2d, 0xa9, 0xe9, 0xb4, 0xed, 0xac, 0x32, 0x12, 0x57, 0xae, 0x74, 0xe0, 0xcc, 0x27, 0xc4, 0x91,
+	0xd7, 0xca, 0xbc, 0x1d, 0xbd, 0xae, 0xbf, 0xd2, 0x20, 0xbb, 0x33, 0x9b, 0x30, 0x8f, 0xcf, 0x29,
+	0x9a, 0x80, 0x11, 0x9b, 0x57, 0xb9, 0x3a, 0xee, 0x5f, 0x6d, 0xa8, 0xb9, 0x6d, 0xf1, 0xee, 0xd5,
+	0xf6, 0x63, 0x25, 0x51, 0xd5, 0x1e, 0x68, 0xe8, 0x29, 0xe4, 0xb9, 0x32, 0x2a, 0x3d, 0xaa, 0xbc,
+	0xbf, 0xed, 0x8a, 0xb7, 0xaf, 0xc0, 0x9d, 0x84, 0x6f, 0xde, 0x3b, 0x7e, 0x5d, 0x4a, 0x1c, 0x9f,
+	0x96, 0xb4, 0x97, 0xa7, 0x25, 0xed, 0xd5, 0x69, 0x49, 0xfb, 0xe3, 0xb4, 0xa4, 0xfd, 0xf4, 0xa6,
+	0x94, 0x78, 0xf9, 0xa6, 0x94, 0x78, 0xf5, 0xa6, 0x94, 0xf8, 0x3e, 0xad, 0x10, 0xfe, 0x0a, 0x00,
+	0x00, 0xff, 0xff, 0xd1, 0xe2, 0x72, 0x64, 0x9c, 0x0c, 0x00, 0x00,
 }
diff --git a/pkg/storage/raft.proto b/pkg/storage/raft.proto
index b008a8cada9f..3cad2bc76021 100644
--- a/pkg/storage/raft.proto
+++ b/pkg/storage/raft.proto
@@ -35,6 +35,16 @@ message RaftHeartbeat {
   optional uint64 term = 4 [(gogoproto.nullable) = false];
   optional uint64 commit = 5 [(gogoproto.nullable) = false];
   optional bool quiesce = 6 [(gogoproto.nullable) = false];
+
+  // ToIsLearner was added in v19.2 to aid in the transition from preemptive
+  // snapshots to learner replicas. If a Replica learns its ID from a message
+  // which indicates that it is a learner and it is not currently a part of the
+  // range (due to being from a preemptive snapshot) then it must delete all of
+  // its data.
+  //
+  // TODO(ajwerner): remove in 20.2 once we ensure that preemptive snapshots can
+  // no longer be present and that we're never talking to a 19.2 node.
+  optional bool to_is_learner = 7 [(gogoproto.nullable) = false];
 }
 
 // RaftMessageRequest is the request used to send raft messages using our
diff --git a/pkg/storage/replica_application_state_machine.go b/pkg/storage/replica_application_state_machine.go
index a6b83c3c846b..605b15fa99f9 100644
--- a/pkg/storage/replica_application_state_machine.go
+++ b/pkg/storage/replica_application_state_machine.go
@@ -439,7 +439,25 @@ func (b *replicaAppBatch) Stage(cmdI apply.Command) (apply.CheckedCommand, error
 		log.Event(ctx, "applying command")
 	}
 
-	// Acquire the split or merge lock, if necessary. If a split or merge
+	// There's a nasty edge case here which only exists in 19.2 (for subtle
+	// reasons).
+	//
+	// Imagine we're catching up from a pre-emptive snapshot and we come across
+	// a merge, we're not gaurenteed that the RHS is going to be present. One
+	// option might be to destroy the current replica. Unfortunately we might
+	// also gotten raft messages and voted so we'll know our replica ID but our
+	// current view of the range descriptor will not include the current store.
+	//
+	// * One proposal is to just refuse to apply the current command and just spin
+	//   handling raft ready objects but not actually applying any commands but
+	//   detecting the illegal merge and truncating the CommittedEntries slice
+	//   to before the merge prior to calling Advance.
+	// * The other is to permit destroying an initialized replica with
+	//   nextReplicaID == Replica.mu.replicaID if the store is not in
+	//   Replica.mu.state.Desc. We'd need to make sure to preserve the hard state
+	//   in this case.
+
+	// acquire the split or merge lock, if necessary. If a split or merge
 	// command was rejected with a below-Raft forced error then its replicated
 	// result was just cleared and this will be a no-op.
 	if splitMergeUnlock, err := b.r.maybeAcquireSplitMergeLock(ctx, cmd.raftCmd); err != nil {
diff --git a/pkg/storage/replica_raft.go b/pkg/storage/replica_raft.go
index aa68401ea88a..64b32169203a 100644
--- a/pkg/storage/replica_raft.go
+++ b/pkg/storage/replica_raft.go
@@ -1048,6 +1048,7 @@ func (r *Replica) maybeCoalesceHeartbeat(
 		Term:          msg.Term,
 		Commit:        msg.Commit,
 		Quiesce:       quiesce,
+		ToIsLearner:   toReplica.GetType() == roachpb.LEARNER,
 	}
 	if log.V(4) {
 		log.Infof(ctx, "coalescing beat: %+v", beat)
@@ -1572,7 +1573,7 @@ func (r *Replica) maybeAcquireSplitMergeLock(
 func (r *Replica) acquireSplitLock(
 	ctx context.Context, split *roachpb.SplitTrigger,
 ) (func(), error) {
-	rightRng, _, err := r.store.getOrCreateReplica(ctx, split.RightDesc.RangeID, 0, nil)
+	rightRng, _, err := r.store.getOrCreateReplica(ctx, split.RightDesc.RangeID, 0, nil, true)
 	if err != nil {
 		return nil, err
 	}
@@ -1594,7 +1595,13 @@ func (r *Replica) acquireMergeLock(
 	// right-hand replica in place, any snapshots for the right-hand range will
 	// block on raftMu, waiting for the merge to complete, after which the replica
 	// will realize it has been destroyed and reject the snapshot.
-	rightRepl, _, err := r.store.getOrCreateReplica(ctx, merge.RightDesc.RangeID, 0, nil)
+	//
+	// There's one edge case to this requirement: the current Replica may be
+	// catching up from a pre-emptive snapshot. This case only arises in v19.2
+	// while performing the upgrade. If we detect that we're not a part of this
+	// range then it's not safe to apply this merge. In that case we need to
+	// remove the LHS replica synchronously.
+	rightRepl, _, err := r.store.getOrCreateReplica(ctx, merge.RightDesc.RangeID, 0, nil, true)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/storage/store.go b/pkg/storage/store.go
index 068270f9bcc6..b415c91d7a6a 100644
--- a/pkg/storage/store.go
+++ b/pkg/storage/store.go
@@ -3356,6 +3356,10 @@ func (s *Store) HandleSnapshot(
 	})
 }
 
+// learnerReplicaType exists to avoid allocating when marking a replica as a
+// learner in uncoalesceBeats.
+var learnerReplicaType = roachpb.LEARNER
+
 func (s *Store) uncoalesceBeats(
 	ctx context.Context,
 	beats []RaftHeartbeat,
@@ -3393,6 +3397,9 @@ func (s *Store) uncoalesceBeats(
 			Message: msg,
 			Quiesce: beat.Quiesce,
 		}
+		if beat.ToIsLearner {
+			beatReqs[i].ToReplica.Type = &learnerReplicaType
+		}
 		if log.V(4) {
 			log.Infof(ctx, "uncoalesced beat: %+v", beatReqs[i])
 		}
@@ -3479,6 +3486,7 @@ func (s *Store) withReplicaForRequest(
 		req.RangeID,
 		req.ToReplica.ReplicaID,
 		&req.FromReplica,
+		req.ToReplica.GetType() == roachpb.VOTER_FULL,
 	)
 	if err != nil {
 		return roachpb.NewError(err)
@@ -4063,6 +4071,7 @@ func (s *Store) getOrCreateReplica(
 	rangeID roachpb.RangeID,
 	replicaID roachpb.ReplicaID,
 	creatingReplica *roachpb.ReplicaDescriptor,
+	isVoter bool,
 ) (_ *Replica, created bool, _ error) {
 	r := retry.Start(retry.Options{
 		InitialBackoff: time.Microsecond,
@@ -4075,6 +4084,7 @@ func (s *Store) getOrCreateReplica(
 			rangeID,
 			replicaID,
 			creatingReplica,
+			isVoter,
 		)
 		if err == errRetry {
 			continue
@@ -4097,8 +4107,27 @@ func (s *Store) tryGetOrCreateReplica(
 	rangeID roachpb.RangeID,
 	replicaID roachpb.ReplicaID,
 	creatingReplica *roachpb.ReplicaDescriptor,
+	isVoter bool,
 ) (_ *Replica, created bool, err error) {
 	var (
+		removeReplica = func(repl *Replica) error {
+			repl.mu.destroyStatus.Set(err, destroyReasonRemovalPending)
+			isInitialized := repl.isInitializedRLocked()
+			repl.mu.Unlock()
+			defer repl.mu.Lock()
+			if !isInitialized {
+				if err := s.removeUninitializedReplicaRaftMuLocked(ctx, repl, replicaID); err != nil {
+					log.Fatalf(ctx, "failed to remove uninitialized replica: %v", err)
+				}
+			} else {
+				if err := s.removeReplicaImpl(ctx, repl, replicaID, RemoveOptions{
+					DestroyData: true,
+				}); err != nil {
+					log.Fatal(ctx, err)
+				}
+			}
+			return errRetry
+		}
 		handleFromReplicaTooOld = func(repl *Replica) error {
 			if creatingReplica == nil {
 				return nil
@@ -4120,22 +4149,16 @@ func (s *Store) tryGetOrCreateReplica(
 			if log.V(1) {
 				log.Infof(ctx, "found message for newer replica ID: %v %v %v %v %v", repl.mu.replicaID, replicaID, repl.mu.minReplicaID, repl.mu.state.Desc, &repl.mu.destroyStatus)
 			}
-			repl.mu.destroyStatus.Set(err, destroyReasonRemovalPending)
-			isInitialized := repl.isInitializedRLocked()
-			repl.mu.Unlock()
-			defer repl.mu.Lock()
-			if !isInitialized {
-				if err := s.removeUninitializedReplicaRaftMuLocked(ctx, repl, replicaID); err != nil {
-					log.Fatalf(ctx, "failed to remove uninitialized replica: %v", err)
-				}
-			} else {
-				if err := s.removeReplicaImpl(ctx, repl, replicaID, RemoveOptions{
-					DestroyData: true,
-				}); err != nil {
-					log.Fatal(ctx, err)
-				}
+			return removeReplica(repl)
+		}
+		handleNonVoterWithPreemptiveSnapshot = func(repl *Replica) error {
+			if repl.mu.replicaID != 0 || replicaID == 0 || !repl.isInitializedRLocked() || isVoter {
+				return nil
 			}
-			return errRetry
+			if log.V(1) {
+				log.Infof(ctx, "found message for replica ID %v as non-voter but currently not part of the range, destroying preemptive snapshot", replicaID)
+			}
+			return removeReplica(repl)
 		}
 	)
 	// The common case: look up an existing (initialized) replica.
@@ -4156,11 +4179,16 @@ func (s *Store) tryGetOrCreateReplica(
 			repl.raftMu.Unlock()
 			return nil, false, err
 		}
+		if err := handleNonVoterWithPreemptiveSnapshot(repl); err != nil {
+			repl.raftMu.Unlock()
+			return nil, false, err
+		}
 
 		var err error
-		if repl.mu.replicaID == 0 {
+		if repl.mu.replicaID == 0 && replicaID != 0 {
 			// This message is telling us about our replica ID.
-			// This is a common case when dealing with preemptive snapshots.
+			// This is a common case when dealing with preemptive snapshots if we're
+			// initialized.
 			err = repl.setReplicaIDRaftMuLockedMuLocked(repl.AnnotateCtx(ctx), replicaID)
 		} else if replicaID != 0 && repl.mu.replicaID > replicaID {
 			// TODO(ajwerner): probably just silently drop this message.
diff --git a/pkg/storage/store_snapshot.go b/pkg/storage/store_snapshot.go
index 2674ed25f2e9..9c138f8846ec 100644
--- a/pkg/storage/store_snapshot.go
+++ b/pkg/storage/store_snapshot.go
@@ -744,11 +744,19 @@ func (s *Store) shouldAcceptSnapshotData(
 	}
 	pErr := s.withReplicaForRequest(ctx, &snapHeader.RaftMessageRequest,
 		func(ctx context.Context, r *Replica) *roachpb.Error {
+			// If the current replica is not initialized then we should accept this
+			// snapshot if it doesn't overlap existing ranges.
 			if !r.IsInitialized() {
 				s.mu.Lock()
 				defer s.mu.Unlock()
 				return roachpb.NewError(s.checkSnapshotOverlapLocked(ctx, snapHeader))
 			}
+			// If the current range is initialized then we need to accept this
+			// this snapshot. There's a hidden nasty case here during 19.2 where
+			// our currently initialized range is due to a preemptive snapshot and
+			// we've now assigned that range a replica ID. Fundamentally at this
+			// point we want to clear out the pre-emptive snapshot because applying
+			// learner snapshot over top is likely going to be problematic.
 			return nil
 		})
 	return pErr.GoError()
diff --git a/pkg/storage/store_snapshot_test.go b/pkg/storage/store_snapshot_test.go
index 12c10f2e7316..74f3aaef5cde 100644
--- a/pkg/storage/store_snapshot_test.go
+++ b/pkg/storage/store_snapshot_test.go
@@ -114,7 +114,7 @@ func TestSnapshotPreemptiveOnUninitializedReplica(t *testing.T) {
 	store, _ := createTestStore(t, testStoreOpts{}, stopper)
 
 	// Create an uninitialized replica.
-	repl, created, err := store.getOrCreateReplica(ctx, 77, 1, nil)
+	repl, created, err := store.getOrCreateReplica(ctx, 77, 1, nil, true)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -141,3 +141,15 @@ func TestSnapshotPreemptiveOnUninitializedReplica(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+// TestSnapshotPreemptiveRemovedAfterMessageToNonVoter tests that replicas which
+// encounter preemptive snapshots remove all of the data due to that preemptive
+// snapshot.
+//
+// This behavior is critical to provide safety when upgrading from preemptive
+// snapshots. Before learner replicas we would not add a store to a range
+// until we had successfully sent a preemptive snapshot. Furthermore we'd
+// only add that store assuming that the
+func TestPreemptiveSnapshotRemovedAfterMessageTo(t *testing.T) {
+
+}