node_test.go 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789
  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. package raft
  15. import (
  16. "bytes"
  17. "context"
  18. "reflect"
  19. "testing"
  20. "time"
  21. "github.com/coreos/etcd/pkg/testutil"
  22. "github.com/coreos/etcd/raft/raftpb"
  23. )
  24. // TestNodeStep ensures that node.Step sends msgProp to propc chan
  25. // and other kinds of messages to recvc chan.
  26. func TestNodeStep(t *testing.T) {
  27. for i, msgn := range raftpb.MessageType_name {
  28. n := &node{
  29. propc: make(chan raftpb.Message, 1),
  30. recvc: make(chan raftpb.Message, 1),
  31. }
  32. msgt := raftpb.MessageType(i)
  33. n.Step(context.TODO(), raftpb.Message{Type: msgt})
  34. // Proposal goes to proc chan. Others go to recvc chan.
  35. if msgt == raftpb.MsgProp {
  36. select {
  37. case <-n.propc:
  38. default:
  39. t.Errorf("%d: cannot receive %s on propc chan", msgt, msgn)
  40. }
  41. } else {
  42. if IsLocalMsg(msgt) {
  43. select {
  44. case <-n.recvc:
  45. t.Errorf("%d: step should ignore %s", msgt, msgn)
  46. default:
  47. }
  48. } else {
  49. select {
  50. case <-n.recvc:
  51. default:
  52. t.Errorf("%d: cannot receive %s on recvc chan", msgt, msgn)
  53. }
  54. }
  55. }
  56. }
  57. }
  58. // Cancel and Stop should unblock Step()
  59. func TestNodeStepUnblock(t *testing.T) {
  60. // a node without buffer to block step
  61. n := &node{
  62. propc: make(chan raftpb.Message),
  63. done: make(chan struct{}),
  64. }
  65. ctx, cancel := context.WithCancel(context.Background())
  66. stopFunc := func() { close(n.done) }
  67. tests := []struct {
  68. unblock func()
  69. werr error
  70. }{
  71. {stopFunc, ErrStopped},
  72. {cancel, context.Canceled},
  73. }
  74. for i, tt := range tests {
  75. errc := make(chan error, 1)
  76. go func() {
  77. err := n.Step(ctx, raftpb.Message{Type: raftpb.MsgProp})
  78. errc <- err
  79. }()
  80. tt.unblock()
  81. select {
  82. case err := <-errc:
  83. if err != tt.werr {
  84. t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
  85. }
  86. //clean up side-effect
  87. if ctx.Err() != nil {
  88. ctx = context.TODO()
  89. }
  90. select {
  91. case <-n.done:
  92. n.done = make(chan struct{})
  93. default:
  94. }
  95. case <-time.After(1 * time.Second):
  96. t.Fatalf("#%d: failed to unblock step", i)
  97. }
  98. }
  99. }
  100. // TestNodePropose ensures that node.Propose sends the given proposal to the underlying raft.
  101. func TestNodePropose(t *testing.T) {
  102. msgs := []raftpb.Message{}
  103. appendStep := func(r *raft, m raftpb.Message) error {
  104. msgs = append(msgs, m)
  105. return nil
  106. }
  107. n := newNode()
  108. s := NewMemoryStorage()
  109. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  110. go n.run(r)
  111. n.Campaign(context.TODO())
  112. for {
  113. rd := <-n.Ready()
  114. s.Append(rd.Entries)
  115. // change the step function to appendStep until this raft becomes leader
  116. if rd.SoftState.Lead == r.id {
  117. r.step = appendStep
  118. n.Advance()
  119. break
  120. }
  121. n.Advance()
  122. }
  123. n.Propose(context.TODO(), []byte("somedata"))
  124. n.Stop()
  125. if len(msgs) != 1 {
  126. t.Fatalf("len(msgs) = %d, want %d", len(msgs), 1)
  127. }
  128. if msgs[0].Type != raftpb.MsgProp {
  129. t.Errorf("msg type = %d, want %d", msgs[0].Type, raftpb.MsgProp)
  130. }
  131. if !bytes.Equal(msgs[0].Entries[0].Data, []byte("somedata")) {
  132. t.Errorf("data = %v, want %v", msgs[0].Entries[0].Data, []byte("somedata"))
  133. }
  134. }
  135. // TestNodeReadIndex ensures that node.ReadIndex sends the MsgReadIndex message to the underlying raft.
  136. // It also ensures that ReadState can be read out through ready chan.
  137. func TestNodeReadIndex(t *testing.T) {
  138. msgs := []raftpb.Message{}
  139. appendStep := func(r *raft, m raftpb.Message) error {
  140. msgs = append(msgs, m)
  141. return nil
  142. }
  143. wrs := []ReadState{{Index: uint64(1), RequestCtx: []byte("somedata")}}
  144. n := newNode()
  145. s := NewMemoryStorage()
  146. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  147. r.readStates = wrs
  148. go n.run(r)
  149. n.Campaign(context.TODO())
  150. for {
  151. rd := <-n.Ready()
  152. if !reflect.DeepEqual(rd.ReadStates, wrs) {
  153. t.Errorf("ReadStates = %v, want %v", rd.ReadStates, wrs)
  154. }
  155. s.Append(rd.Entries)
  156. if rd.SoftState.Lead == r.id {
  157. n.Advance()
  158. break
  159. }
  160. n.Advance()
  161. }
  162. r.step = appendStep
  163. wrequestCtx := []byte("somedata2")
  164. n.ReadIndex(context.TODO(), wrequestCtx)
  165. n.Stop()
  166. if len(msgs) != 1 {
  167. t.Fatalf("len(msgs) = %d, want %d", len(msgs), 1)
  168. }
  169. if msgs[0].Type != raftpb.MsgReadIndex {
  170. t.Errorf("msg type = %d, want %d", msgs[0].Type, raftpb.MsgReadIndex)
  171. }
  172. if !bytes.Equal(msgs[0].Entries[0].Data, wrequestCtx) {
  173. t.Errorf("data = %v, want %v", msgs[0].Entries[0].Data, wrequestCtx)
  174. }
  175. }
  176. // TestDisableProposalForwarding ensures that proposals are not forwarded to
  177. // the leader when DisableProposalForwarding is true.
  178. func TestDisableProposalForwarding(t *testing.T) {
  179. r1 := newTestRaft(1, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage())
  180. r2 := newTestRaft(2, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage())
  181. cfg3 := newTestConfig(3, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage())
  182. cfg3.DisableProposalForwarding = true
  183. r3 := newRaft(cfg3)
  184. nt := newNetwork(r1, r2, r3)
  185. // elect r1 as leader
  186. nt.send(raftpb.Message{From: 1, To: 1, Type: raftpb.MsgHup})
  187. var testEntries = []raftpb.Entry{{Data: []byte("testdata")}}
  188. // send proposal to r2(follower) where DisableProposalForwarding is false
  189. r2.Step(raftpb.Message{From: 2, To: 2, Type: raftpb.MsgProp, Entries: testEntries})
  190. // verify r2(follower) does forward the proposal when DisableProposalForwarding is false
  191. if len(r2.msgs) != 1 {
  192. t.Fatalf("len(r2.msgs) expected 1, got %d", len(r2.msgs))
  193. }
  194. // send proposal to r3(follower) where DisableProposalForwarding is true
  195. r3.Step(raftpb.Message{From: 3, To: 3, Type: raftpb.MsgProp, Entries: testEntries})
  196. // verify r3(follower) does not forward the proposal when DisableProposalForwarding is true
  197. if len(r3.msgs) != 0 {
  198. t.Fatalf("len(r3.msgs) expected 0, got %d", len(r3.msgs))
  199. }
  200. }
  201. // TestNodeReadIndexToOldLeader ensures that raftpb.MsgReadIndex to old leader
  202. // gets forwarded to the new leader and 'send' method does not attach its term.
  203. func TestNodeReadIndexToOldLeader(t *testing.T) {
  204. r1 := newTestRaft(1, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage())
  205. r2 := newTestRaft(2, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage())
  206. r3 := newTestRaft(3, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage())
  207. nt := newNetwork(r1, r2, r3)
  208. // elect r1 as leader
  209. nt.send(raftpb.Message{From: 1, To: 1, Type: raftpb.MsgHup})
  210. var testEntries = []raftpb.Entry{{Data: []byte("testdata")}}
  211. // send readindex request to r2(follower)
  212. r2.Step(raftpb.Message{From: 2, To: 2, Type: raftpb.MsgReadIndex, Entries: testEntries})
  213. // verify r2(follower) forwards this message to r1(leader) with term not set
  214. if len(r2.msgs) != 1 {
  215. t.Fatalf("len(r2.msgs) expected 1, got %d", len(r2.msgs))
  216. }
  217. readIndxMsg1 := raftpb.Message{From: 2, To: 1, Type: raftpb.MsgReadIndex, Entries: testEntries}
  218. if !reflect.DeepEqual(r2.msgs[0], readIndxMsg1) {
  219. t.Fatalf("r2.msgs[0] expected %+v, got %+v", readIndxMsg1, r2.msgs[0])
  220. }
  221. // send readindex request to r3(follower)
  222. r3.Step(raftpb.Message{From: 3, To: 3, Type: raftpb.MsgReadIndex, Entries: testEntries})
  223. // verify r3(follower) forwards this message to r1(leader) with term not set as well.
  224. if len(r3.msgs) != 1 {
  225. t.Fatalf("len(r3.msgs) expected 1, got %d", len(r3.msgs))
  226. }
  227. readIndxMsg2 := raftpb.Message{From: 3, To: 1, Type: raftpb.MsgReadIndex, Entries: testEntries}
  228. if !reflect.DeepEqual(r3.msgs[0], readIndxMsg2) {
  229. t.Fatalf("r3.msgs[0] expected %+v, got %+v", readIndxMsg2, r3.msgs[0])
  230. }
  231. // now elect r3 as leader
  232. nt.send(raftpb.Message{From: 3, To: 3, Type: raftpb.MsgHup})
  233. // let r1 steps the two messages previously we got from r2, r3
  234. r1.Step(readIndxMsg1)
  235. r1.Step(readIndxMsg2)
  236. // verify r1(follower) forwards these messages again to r3(new leader)
  237. if len(r1.msgs) != 2 {
  238. t.Fatalf("len(r1.msgs) expected 1, got %d", len(r1.msgs))
  239. }
  240. readIndxMsg3 := raftpb.Message{From: 1, To: 3, Type: raftpb.MsgReadIndex, Entries: testEntries}
  241. if !reflect.DeepEqual(r1.msgs[0], readIndxMsg3) {
  242. t.Fatalf("r1.msgs[0] expected %+v, got %+v", readIndxMsg3, r1.msgs[0])
  243. }
  244. if !reflect.DeepEqual(r1.msgs[1], readIndxMsg3) {
  245. t.Fatalf("r1.msgs[1] expected %+v, got %+v", readIndxMsg3, r1.msgs[1])
  246. }
  247. }
  248. // TestNodeProposeConfig ensures that node.ProposeConfChange sends the given configuration proposal
  249. // to the underlying raft.
  250. func TestNodeProposeConfig(t *testing.T) {
  251. msgs := []raftpb.Message{}
  252. appendStep := func(r *raft, m raftpb.Message) error {
  253. msgs = append(msgs, m)
  254. return nil
  255. }
  256. n := newNode()
  257. s := NewMemoryStorage()
  258. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  259. go n.run(r)
  260. n.Campaign(context.TODO())
  261. for {
  262. rd := <-n.Ready()
  263. s.Append(rd.Entries)
  264. // change the step function to appendStep until this raft becomes leader
  265. if rd.SoftState.Lead == r.id {
  266. r.step = appendStep
  267. n.Advance()
  268. break
  269. }
  270. n.Advance()
  271. }
  272. cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
  273. ccdata, err := cc.Marshal()
  274. if err != nil {
  275. t.Fatal(err)
  276. }
  277. n.ProposeConfChange(context.TODO(), cc)
  278. n.Stop()
  279. if len(msgs) != 1 {
  280. t.Fatalf("len(msgs) = %d, want %d", len(msgs), 1)
  281. }
  282. if msgs[0].Type != raftpb.MsgProp {
  283. t.Errorf("msg type = %d, want %d", msgs[0].Type, raftpb.MsgProp)
  284. }
  285. if !bytes.Equal(msgs[0].Entries[0].Data, ccdata) {
  286. t.Errorf("data = %v, want %v", msgs[0].Entries[0].Data, ccdata)
  287. }
  288. }
  289. // TestNodeProposeAddDuplicateNode ensures that two proposes to add the same node should
  290. // not affect the later propose to add new node.
  291. func TestNodeProposeAddDuplicateNode(t *testing.T) {
  292. n := newNode()
  293. s := NewMemoryStorage()
  294. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  295. go n.run(r)
  296. n.Campaign(context.TODO())
  297. rdyEntries := make([]raftpb.Entry, 0)
  298. ticker := time.NewTicker(time.Millisecond * 100)
  299. defer ticker.Stop()
  300. done := make(chan struct{})
  301. stop := make(chan struct{})
  302. applyConfChan := make(chan struct{})
  303. go func() {
  304. defer close(done)
  305. for {
  306. select {
  307. case <-stop:
  308. return
  309. case <-ticker.C:
  310. n.Tick()
  311. case rd := <-n.Ready():
  312. s.Append(rd.Entries)
  313. applied := false
  314. for _, e := range rd.Entries {
  315. rdyEntries = append(rdyEntries, e)
  316. switch e.Type {
  317. case raftpb.EntryNormal:
  318. case raftpb.EntryConfChange:
  319. var cc raftpb.ConfChange
  320. cc.Unmarshal(e.Data)
  321. n.ApplyConfChange(cc)
  322. applied = true
  323. }
  324. }
  325. n.Advance()
  326. if applied {
  327. applyConfChan <- struct{}{}
  328. }
  329. }
  330. }
  331. }()
  332. cc1 := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
  333. ccdata1, _ := cc1.Marshal()
  334. n.ProposeConfChange(context.TODO(), cc1)
  335. <-applyConfChan
  336. // try add the same node again
  337. n.ProposeConfChange(context.TODO(), cc1)
  338. <-applyConfChan
  339. // the new node join should be ok
  340. cc2 := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 2}
  341. ccdata2, _ := cc2.Marshal()
  342. n.ProposeConfChange(context.TODO(), cc2)
  343. <-applyConfChan
  344. close(stop)
  345. <-done
  346. if len(rdyEntries) != 4 {
  347. t.Errorf("len(entry) = %d, want %d, %v\n", len(rdyEntries), 4, rdyEntries)
  348. }
  349. if !bytes.Equal(rdyEntries[1].Data, ccdata1) {
  350. t.Errorf("data = %v, want %v", rdyEntries[1].Data, ccdata1)
  351. }
  352. if !bytes.Equal(rdyEntries[3].Data, ccdata2) {
  353. t.Errorf("data = %v, want %v", rdyEntries[3].Data, ccdata2)
  354. }
  355. n.Stop()
  356. }
  357. // TestBlockProposal ensures that node will block proposal when it does not
  358. // know who is the current leader; node will accept proposal when it knows
  359. // who is the current leader.
  360. func TestBlockProposal(t *testing.T) {
  361. n := newNode()
  362. r := newTestRaft(1, []uint64{1}, 10, 1, NewMemoryStorage())
  363. go n.run(r)
  364. defer n.Stop()
  365. errc := make(chan error, 1)
  366. go func() {
  367. errc <- n.Propose(context.TODO(), []byte("somedata"))
  368. }()
  369. testutil.WaitSchedule()
  370. select {
  371. case err := <-errc:
  372. t.Errorf("err = %v, want blocking", err)
  373. default:
  374. }
  375. n.Campaign(context.TODO())
  376. select {
  377. case err := <-errc:
  378. if err != nil {
  379. t.Errorf("err = %v, want %v", err, nil)
  380. }
  381. case <-time.After(10 * time.Second):
  382. t.Errorf("blocking proposal, want unblocking")
  383. }
  384. }
  385. // TestNodeTick ensures that node.Tick() will increase the
  386. // elapsed of the underlying raft state machine.
  387. func TestNodeTick(t *testing.T) {
  388. n := newNode()
  389. s := NewMemoryStorage()
  390. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  391. go n.run(r)
  392. elapsed := r.electionElapsed
  393. n.Tick()
  394. for len(n.tickc) != 0 {
  395. time.Sleep(100 * time.Millisecond)
  396. }
  397. n.Stop()
  398. if r.electionElapsed != elapsed+1 {
  399. t.Errorf("elapsed = %d, want %d", r.electionElapsed, elapsed+1)
  400. }
  401. }
  402. // TestNodeStop ensures that node.Stop() blocks until the node has stopped
  403. // processing, and that it is idempotent
  404. func TestNodeStop(t *testing.T) {
  405. n := newNode()
  406. s := NewMemoryStorage()
  407. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  408. donec := make(chan struct{})
  409. go func() {
  410. n.run(r)
  411. close(donec)
  412. }()
  413. status := n.Status()
  414. n.Stop()
  415. select {
  416. case <-donec:
  417. case <-time.After(time.Second):
  418. t.Fatalf("timed out waiting for node to stop!")
  419. }
  420. emptyStatus := Status{}
  421. if reflect.DeepEqual(status, emptyStatus) {
  422. t.Errorf("status = %v, want not empty", status)
  423. }
  424. // Further status should return be empty, the node is stopped.
  425. status = n.Status()
  426. if !reflect.DeepEqual(status, emptyStatus) {
  427. t.Errorf("status = %v, want empty", status)
  428. }
  429. // Subsequent Stops should have no effect.
  430. n.Stop()
  431. }
  432. func TestReadyContainUpdates(t *testing.T) {
  433. tests := []struct {
  434. rd Ready
  435. wcontain bool
  436. }{
  437. {Ready{}, false},
  438. {Ready{SoftState: &SoftState{Lead: 1}}, true},
  439. {Ready{HardState: raftpb.HardState{Vote: 1}}, true},
  440. {Ready{Entries: make([]raftpb.Entry, 1)}, true},
  441. {Ready{CommittedEntries: make([]raftpb.Entry, 1)}, true},
  442. {Ready{Messages: make([]raftpb.Message, 1)}, true},
  443. {Ready{Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1}}}, true},
  444. }
  445. for i, tt := range tests {
  446. if g := tt.rd.containsUpdates(); g != tt.wcontain {
  447. t.Errorf("#%d: containUpdates = %v, want %v", i, g, tt.wcontain)
  448. }
  449. }
  450. }
  451. // TestNodeStart ensures that a node can be started correctly. The node should
  452. // start with correct configuration change entries, and can accept and commit
  453. // proposals.
  454. func TestNodeStart(t *testing.T) {
  455. ctx, cancel := context.WithCancel(context.Background())
  456. defer cancel()
  457. cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
  458. ccdata, err := cc.Marshal()
  459. if err != nil {
  460. t.Fatalf("unexpected marshal error: %v", err)
  461. }
  462. wants := []Ready{
  463. {
  464. HardState: raftpb.HardState{Term: 1, Commit: 1, Vote: 0},
  465. Entries: []raftpb.Entry{
  466. {Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
  467. },
  468. CommittedEntries: []raftpb.Entry{
  469. {Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
  470. },
  471. MustSync: true,
  472. },
  473. {
  474. HardState: raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
  475. Entries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
  476. CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
  477. MustSync: true,
  478. },
  479. }
  480. storage := NewMemoryStorage()
  481. c := &Config{
  482. ID: 1,
  483. ElectionTick: 10,
  484. HeartbeatTick: 1,
  485. Storage: storage,
  486. MaxSizePerMsg: noLimit,
  487. MaxInflightMsgs: 256,
  488. }
  489. n := StartNode(c, []Peer{{ID: 1}})
  490. defer n.Stop()
  491. g := <-n.Ready()
  492. if !reflect.DeepEqual(g, wants[0]) {
  493. t.Fatalf("#%d: g = %+v,\n w %+v", 1, g, wants[0])
  494. } else {
  495. storage.Append(g.Entries)
  496. n.Advance()
  497. }
  498. n.Campaign(ctx)
  499. rd := <-n.Ready()
  500. storage.Append(rd.Entries)
  501. n.Advance()
  502. n.Propose(ctx, []byte("foo"))
  503. if g2 := <-n.Ready(); !reflect.DeepEqual(g2, wants[1]) {
  504. t.Errorf("#%d: g = %+v,\n w %+v", 2, g2, wants[1])
  505. } else {
  506. storage.Append(g2.Entries)
  507. n.Advance()
  508. }
  509. select {
  510. case rd := <-n.Ready():
  511. t.Errorf("unexpected Ready: %+v", rd)
  512. case <-time.After(time.Millisecond):
  513. }
  514. }
  515. func TestNodeRestart(t *testing.T) {
  516. entries := []raftpb.Entry{
  517. {Term: 1, Index: 1},
  518. {Term: 1, Index: 2, Data: []byte("foo")},
  519. }
  520. st := raftpb.HardState{Term: 1, Commit: 1}
  521. want := Ready{
  522. HardState: st,
  523. // commit up to index commit index in st
  524. CommittedEntries: entries[:st.Commit],
  525. MustSync: true,
  526. }
  527. storage := NewMemoryStorage()
  528. storage.SetHardState(st)
  529. storage.Append(entries)
  530. c := &Config{
  531. ID: 1,
  532. ElectionTick: 10,
  533. HeartbeatTick: 1,
  534. Storage: storage,
  535. MaxSizePerMsg: noLimit,
  536. MaxInflightMsgs: 256,
  537. }
  538. n := RestartNode(c)
  539. defer n.Stop()
  540. if g := <-n.Ready(); !reflect.DeepEqual(g, want) {
  541. t.Errorf("g = %+v,\n w %+v", g, want)
  542. }
  543. n.Advance()
  544. select {
  545. case rd := <-n.Ready():
  546. t.Errorf("unexpected Ready: %+v", rd)
  547. case <-time.After(time.Millisecond):
  548. }
  549. }
  550. func TestNodeRestartFromSnapshot(t *testing.T) {
  551. snap := raftpb.Snapshot{
  552. Metadata: raftpb.SnapshotMetadata{
  553. ConfState: raftpb.ConfState{Nodes: []uint64{1, 2}},
  554. Index: 2,
  555. Term: 1,
  556. },
  557. }
  558. entries := []raftpb.Entry{
  559. {Term: 1, Index: 3, Data: []byte("foo")},
  560. }
  561. st := raftpb.HardState{Term: 1, Commit: 3}
  562. want := Ready{
  563. HardState: st,
  564. // commit up to index commit index in st
  565. CommittedEntries: entries,
  566. MustSync: true,
  567. }
  568. s := NewMemoryStorage()
  569. s.SetHardState(st)
  570. s.ApplySnapshot(snap)
  571. s.Append(entries)
  572. c := &Config{
  573. ID: 1,
  574. ElectionTick: 10,
  575. HeartbeatTick: 1,
  576. Storage: s,
  577. MaxSizePerMsg: noLimit,
  578. MaxInflightMsgs: 256,
  579. }
  580. n := RestartNode(c)
  581. defer n.Stop()
  582. if g := <-n.Ready(); !reflect.DeepEqual(g, want) {
  583. t.Errorf("g = %+v,\n w %+v", g, want)
  584. } else {
  585. n.Advance()
  586. }
  587. select {
  588. case rd := <-n.Ready():
  589. t.Errorf("unexpected Ready: %+v", rd)
  590. case <-time.After(time.Millisecond):
  591. }
  592. }
  593. func TestNodeAdvance(t *testing.T) {
  594. ctx, cancel := context.WithCancel(context.Background())
  595. defer cancel()
  596. storage := NewMemoryStorage()
  597. c := &Config{
  598. ID: 1,
  599. ElectionTick: 10,
  600. HeartbeatTick: 1,
  601. Storage: storage,
  602. MaxSizePerMsg: noLimit,
  603. MaxInflightMsgs: 256,
  604. }
  605. n := StartNode(c, []Peer{{ID: 1}})
  606. defer n.Stop()
  607. rd := <-n.Ready()
  608. storage.Append(rd.Entries)
  609. n.Advance()
  610. n.Campaign(ctx)
  611. <-n.Ready()
  612. n.Propose(ctx, []byte("foo"))
  613. select {
  614. case rd = <-n.Ready():
  615. t.Fatalf("unexpected Ready before Advance: %+v", rd)
  616. case <-time.After(time.Millisecond):
  617. }
  618. storage.Append(rd.Entries)
  619. n.Advance()
  620. select {
  621. case <-n.Ready():
  622. case <-time.After(100 * time.Millisecond):
  623. t.Errorf("expect Ready after Advance, but there is no Ready available")
  624. }
  625. }
  626. func TestSoftStateEqual(t *testing.T) {
  627. tests := []struct {
  628. st *SoftState
  629. we bool
  630. }{
  631. {&SoftState{}, true},
  632. {&SoftState{Lead: 1}, false},
  633. {&SoftState{RaftState: StateLeader}, false},
  634. }
  635. for i, tt := range tests {
  636. if g := tt.st.equal(&SoftState{}); g != tt.we {
  637. t.Errorf("#%d, equal = %v, want %v", i, g, tt.we)
  638. }
  639. }
  640. }
  641. func TestIsHardStateEqual(t *testing.T) {
  642. tests := []struct {
  643. st raftpb.HardState
  644. we bool
  645. }{
  646. {emptyState, true},
  647. {raftpb.HardState{Vote: 1}, false},
  648. {raftpb.HardState{Commit: 1}, false},
  649. {raftpb.HardState{Term: 1}, false},
  650. }
  651. for i, tt := range tests {
  652. if isHardStateEqual(tt.st, emptyState) != tt.we {
  653. t.Errorf("#%d, equal = %v, want %v", i, isHardStateEqual(tt.st, emptyState), tt.we)
  654. }
  655. }
  656. }
  657. func TestNodeProposeAddLearnerNode(t *testing.T) {
  658. ticker := time.NewTicker(time.Millisecond * 100)
  659. defer ticker.Stop()
  660. n := newNode()
  661. s := NewMemoryStorage()
  662. r := newTestRaft(1, []uint64{1}, 10, 1, s)
  663. go n.run(r)
  664. n.Campaign(context.TODO())
  665. stop := make(chan struct{})
  666. done := make(chan struct{})
  667. applyConfChan := make(chan struct{})
  668. go func() {
  669. defer close(done)
  670. for {
  671. select {
  672. case <-stop:
  673. return
  674. case <-ticker.C:
  675. n.Tick()
  676. case rd := <-n.Ready():
  677. s.Append(rd.Entries)
  678. t.Logf("raft: %v", rd.Entries)
  679. for _, ent := range rd.Entries {
  680. if ent.Type != raftpb.EntryConfChange {
  681. continue
  682. }
  683. var cc raftpb.ConfChange
  684. cc.Unmarshal(ent.Data)
  685. state := n.ApplyConfChange(cc)
  686. if len(state.Learners) == 0 ||
  687. state.Learners[0] != cc.NodeID ||
  688. cc.NodeID != 2 {
  689. t.Errorf("apply conf change should return new added learner: %v", state.String())
  690. }
  691. if len(state.Nodes) != 1 {
  692. t.Errorf("add learner should not change the nodes: %v", state.String())
  693. }
  694. t.Logf("apply raft conf %v changed to: %v", cc, state.String())
  695. applyConfChan <- struct{}{}
  696. }
  697. n.Advance()
  698. }
  699. }
  700. }()
  701. cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddLearnerNode, NodeID: 2}
  702. n.ProposeConfChange(context.TODO(), cc)
  703. <-applyConfChan
  704. close(stop)
  705. <-done
  706. }