raft_snap_test.go

Documentation: go.etcd.io/etcd/raft/v3

     1  // Copyright 2015 The etcd Authors
     2  //
     3  // Licensed under the Apache License, Version 2.0 (the "License");
     4  // you may not use this file except in compliance with the License.
     5  // You may obtain a copy of the License at
     6  //
     7  //     http://www.apache.org/licenses/LICENSE-2.0
     8  //
     9  // Unless required by applicable law or agreed to in writing, software
    10  // distributed under the License is distributed on an "AS IS" BASIS,
    11  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12  // See the License for the specific language governing permissions and
    13  // limitations under the License.
    14  
    15  package raft
    16  
    17  import (
    18  	"testing"
    19  
    20  	pb "go.etcd.io/etcd/raft/v3/raftpb"
    21  )
    22  
    23  var (
    24  	testingSnap = pb.Snapshot{
    25  		Metadata: pb.SnapshotMetadata{
    26  			Index:     11, // magic number
    27  			Term:      11, // magic number
    28  			ConfState: pb.ConfState{Voters: []uint64{1, 2}},
    29  		},
    30  	}
    31  )
    32  
    33  func TestSendingSnapshotSetPendingSnapshot(t *testing.T) {
    34  	storage := newTestMemoryStorage(withPeers(1))
    35  	sm := newTestRaft(1, 10, 1, storage)
    36  	sm.restore(testingSnap)
    37  
    38  	sm.becomeCandidate()
    39  	sm.becomeLeader()
    40  
    41  	// force set the next of node 2, so that
    42  	// node 2 needs a snapshot
    43  	sm.prs.Progress[2].Next = sm.raftLog.firstIndex()
    44  
    45  	sm.Step(pb.Message{From: 2, To: 1, Type: pb.MsgAppResp, Index: sm.prs.Progress[2].Next - 1, Reject: true})
    46  	if sm.prs.Progress[2].PendingSnapshot != 11 {
    47  		t.Fatalf("PendingSnapshot = %d, want 11", sm.prs.Progress[2].PendingSnapshot)
    48  	}
    49  }
    50  
    51  func TestPendingSnapshotPauseReplication(t *testing.T) {
    52  	storage := newTestMemoryStorage(withPeers(1, 2))
    53  	sm := newTestRaft(1, 10, 1, storage)
    54  	sm.restore(testingSnap)
    55  
    56  	sm.becomeCandidate()
    57  	sm.becomeLeader()
    58  
    59  	sm.prs.Progress[2].BecomeSnapshot(11)
    60  
    61  	sm.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
    62  	msgs := sm.readMessages()
    63  	if len(msgs) != 0 {
    64  		t.Fatalf("len(msgs) = %d, want 0", len(msgs))
    65  	}
    66  }
    67  
    68  func TestSnapshotFailure(t *testing.T) {
    69  	storage := newTestMemoryStorage(withPeers(1, 2))
    70  	sm := newTestRaft(1, 10, 1, storage)
    71  	sm.restore(testingSnap)
    72  
    73  	sm.becomeCandidate()
    74  	sm.becomeLeader()
    75  
    76  	sm.prs.Progress[2].Next = 1
    77  	sm.prs.Progress[2].BecomeSnapshot(11)
    78  
    79  	sm.Step(pb.Message{From: 2, To: 1, Type: pb.MsgSnapStatus, Reject: true})
    80  	if sm.prs.Progress[2].PendingSnapshot != 0 {
    81  		t.Fatalf("PendingSnapshot = %d, want 0", sm.prs.Progress[2].PendingSnapshot)
    82  	}
    83  	if sm.prs.Progress[2].Next != 1 {
    84  		t.Fatalf("Next = %d, want 1", sm.prs.Progress[2].Next)
    85  	}
    86  	if !sm.prs.Progress[2].ProbeSent {
    87  		t.Errorf("ProbeSent = %v, want true", sm.prs.Progress[2].ProbeSent)
    88  	}
    89  }
    90  
    91  func TestSnapshotSucceed(t *testing.T) {
    92  	storage := newTestMemoryStorage(withPeers(1, 2))
    93  	sm := newTestRaft(1, 10, 1, storage)
    94  	sm.restore(testingSnap)
    95  
    96  	sm.becomeCandidate()
    97  	sm.becomeLeader()
    98  
    99  	sm.prs.Progress[2].Next = 1
   100  	sm.prs.Progress[2].BecomeSnapshot(11)
   101  
   102  	sm.Step(pb.Message{From: 2, To: 1, Type: pb.MsgSnapStatus, Reject: false})
   103  	if sm.prs.Progress[2].PendingSnapshot != 0 {
   104  		t.Fatalf("PendingSnapshot = %d, want 0", sm.prs.Progress[2].PendingSnapshot)
   105  	}
   106  	if sm.prs.Progress[2].Next != 12 {
   107  		t.Fatalf("Next = %d, want 12", sm.prs.Progress[2].Next)
   108  	}
   109  	if !sm.prs.Progress[2].ProbeSent {
   110  		t.Errorf("ProbeSent = %v, want true", sm.prs.Progress[2].ProbeSent)
   111  	}
   112  }
   113  
   114  func TestSnapshotAbort(t *testing.T) {
   115  	storage := newTestMemoryStorage(withPeers(1, 2))
   116  	sm := newTestRaft(1, 10, 1, storage)
   117  	sm.restore(testingSnap)
   118  
   119  	sm.becomeCandidate()
   120  	sm.becomeLeader()
   121  
   122  	sm.prs.Progress[2].Next = 1
   123  	sm.prs.Progress[2].BecomeSnapshot(11)
   124  
   125  	// A successful msgAppResp that has a higher/equal index than the
   126  	// pending snapshot should abort the pending snapshot.
   127  	sm.Step(pb.Message{From: 2, To: 1, Type: pb.MsgAppResp, Index: 11})
   128  	if sm.prs.Progress[2].PendingSnapshot != 0 {
   129  		t.Fatalf("PendingSnapshot = %d, want 0", sm.prs.Progress[2].PendingSnapshot)
   130  	}
   131  	// The follower entered StateReplicate and the leader send an append
   132  	// and optimistically updated the progress (so we see 13 instead of 12).
   133  	// There is something to append because the leader appended an empty entry
   134  	// to the log at index 12 when it assumed leadership.
   135  	if sm.prs.Progress[2].Next != 13 {
   136  		t.Fatalf("Next = %d, want 13", sm.prs.Progress[2].Next)
   137  	}
   138  	if n := sm.prs.Progress[2].Inflights.Count(); n != 1 {
   139  		t.Fatalf("expected an inflight message, got %d", n)
   140  	}
   141  }
   142
View as plain text