raft_test.go 27 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033
  1. package raft
  2. import (
  3. "bytes"
  4. "math/rand"
  5. "reflect"
  6. "sort"
  7. "testing"
  8. pb "github.com/coreos/etcd/raft/raftpb"
  9. )
  10. // nextEnts returns the appliable entries and updates the applied index
  11. func (r *raft) nextEnts() (ents []pb.Entry) {
  12. ents = r.raftLog.nextEnts()
  13. r.raftLog.resetNextEnts()
  14. return ents
  15. }
  16. type Interface interface {
  17. Step(m pb.Message) error
  18. ReadMessages() []pb.Message
  19. }
  20. func TestLeaderElection(t *testing.T) {
  21. tests := []struct {
  22. *network
  23. state stateType
  24. }{
  25. {newNetwork(nil, nil, nil), stateLeader},
  26. {newNetwork(nil, nil, nopStepper), stateLeader},
  27. {newNetwork(nil, nopStepper, nopStepper), stateCandidate},
  28. {newNetwork(nil, nopStepper, nopStepper, nil), stateCandidate},
  29. {newNetwork(nil, nopStepper, nopStepper, nil, nil), stateLeader},
  30. // three logs further along than 0
  31. {newNetwork(nil, ents(1), ents(2), ents(1, 3), nil), stateFollower},
  32. // logs converge
  33. {newNetwork(ents(1), nil, ents(2), ents(1), nil), stateLeader},
  34. }
  35. for i, tt := range tests {
  36. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  37. sm := tt.network.peers[0].(*raft)
  38. if sm.state != tt.state {
  39. t.Errorf("#%d: state = %s, want %s", i, sm.state, tt.state)
  40. }
  41. if g := sm.Term; g != 1 {
  42. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  43. }
  44. }
  45. }
  46. func TestLogReplication(t *testing.T) {
  47. tests := []struct {
  48. *network
  49. msgs []pb.Message
  50. wcommitted int64
  51. }{
  52. {
  53. newNetwork(nil, nil, nil),
  54. []pb.Message{
  55. {From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  56. },
  57. 2,
  58. },
  59. {
  60. newNetwork(nil, nil, nil),
  61. []pb.Message{
  62. {From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  63. {From: 0, To: 1, Type: msgHup},
  64. {From: 0, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  65. },
  66. 4,
  67. },
  68. }
  69. for i, tt := range tests {
  70. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  71. for _, m := range tt.msgs {
  72. tt.send(m)
  73. }
  74. for j, x := range tt.network.peers {
  75. sm := x.(*raft)
  76. if sm.raftLog.committed != tt.wcommitted {
  77. t.Errorf("#%d.%d: committed = %d, want %d", i, j, sm.raftLog.committed, tt.wcommitted)
  78. }
  79. ents := make([]pb.Entry, 0)
  80. for _, e := range sm.nextEnts() {
  81. if e.Data != nil {
  82. ents = append(ents, e)
  83. }
  84. }
  85. props := make([]pb.Message, 0)
  86. for _, m := range tt.msgs {
  87. if m.Type == msgProp {
  88. props = append(props, m)
  89. }
  90. }
  91. for k, m := range props {
  92. if !bytes.Equal(ents[k].Data, m.Entries[0].Data) {
  93. t.Errorf("#%d.%d: data = %d, want %d", i, j, ents[k].Data, m.Entries[0].Data)
  94. }
  95. }
  96. }
  97. }
  98. }
  99. func TestSingleNodeCommit(t *testing.T) {
  100. tt := newNetwork(nil)
  101. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  102. tt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  103. tt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  104. sm := tt.peers[0].(*raft)
  105. if sm.raftLog.committed != 3 {
  106. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 3)
  107. }
  108. }
  109. // TestCannotCommitWithoutNewTermEntry tests the entries cannot be committed
  110. // when leader changes, no new proposal comes in and ChangeTerm proposal is
  111. // filtered.
  112. func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
  113. tt := newNetwork(nil, nil, nil, nil, nil)
  114. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  115. // 0 cannot reach 2,3,4
  116. tt.cut(0, 2)
  117. tt.cut(0, 3)
  118. tt.cut(0, 4)
  119. tt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  120. tt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  121. sm := tt.peers[0].(*raft)
  122. if sm.raftLog.committed != 1 {
  123. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  124. }
  125. // network recovery
  126. tt.recover()
  127. // avoid committing ChangeTerm proposal
  128. tt.ignore(msgApp)
  129. // elect 1 as the new leader with term 2
  130. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  131. // no log entries from previous term should be committed
  132. sm = tt.peers[1].(*raft)
  133. if sm.raftLog.committed != 1 {
  134. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  135. }
  136. tt.recover()
  137. // send out a heartbeat
  138. // after append a ChangeTerm entry from the current term, all entries
  139. // should be committed
  140. tt.send(pb.Message{From: 1, To: 1, Type: msgBeat})
  141. if sm.raftLog.committed != 4 {
  142. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4)
  143. }
  144. // still be able to append a entry
  145. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  146. if sm.raftLog.committed != 5 {
  147. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 5)
  148. }
  149. }
  150. // TestCommitWithoutNewTermEntry tests the entries could be committed
  151. // when leader changes, no new proposal comes in.
  152. func TestCommitWithoutNewTermEntry(t *testing.T) {
  153. tt := newNetwork(nil, nil, nil, nil, nil)
  154. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  155. // 0 cannot reach 2,3,4
  156. tt.cut(0, 2)
  157. tt.cut(0, 3)
  158. tt.cut(0, 4)
  159. tt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  160. tt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  161. sm := tt.peers[0].(*raft)
  162. if sm.raftLog.committed != 1 {
  163. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  164. }
  165. // network recovery
  166. tt.recover()
  167. // elect 1 as the new leader with term 2
  168. // after append a ChangeTerm entry from the current term, all entries
  169. // should be committed
  170. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  171. if sm.raftLog.committed != 4 {
  172. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4)
  173. }
  174. }
  175. func TestDuelingCandidates(t *testing.T) {
  176. a := newRaft(0, nil, 0, 0) // k, id are set later
  177. b := newRaft(0, nil, 0, 0)
  178. c := newRaft(0, nil, 0, 0)
  179. nt := newNetwork(a, b, c)
  180. nt.cut(0, 2)
  181. nt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  182. nt.send(pb.Message{From: 2, To: 2, Type: msgHup})
  183. nt.recover()
  184. nt.send(pb.Message{From: 2, To: 2, Type: msgHup})
  185. wlog := &raftLog{ents: []pb.Entry{{}, pb.Entry{Type: Normal, Data: nil, Term: 1, Index: 1}}, committed: 1}
  186. tests := []struct {
  187. sm *raft
  188. state stateType
  189. term int64
  190. raftLog *raftLog
  191. }{
  192. {a, stateFollower, 2, wlog},
  193. {b, stateFollower, 2, wlog},
  194. {c, stateFollower, 2, newLog()},
  195. }
  196. for i, tt := range tests {
  197. if g := tt.sm.state; g != tt.state {
  198. t.Errorf("#%d: state = %s, want %s", i, g, tt.state)
  199. }
  200. if g := tt.sm.Term; g != tt.term {
  201. t.Errorf("#%d: term = %d, want %d", i, g, tt.term)
  202. }
  203. base := ltoa(tt.raftLog)
  204. if sm, ok := nt.peers[int64(i)].(*raft); ok {
  205. l := ltoa(sm.raftLog)
  206. if g := diffu(base, l); g != "" {
  207. t.Errorf("#%d: diff:\n%s", i, g)
  208. }
  209. } else {
  210. t.Logf("#%d: empty log", i)
  211. }
  212. }
  213. }
  214. func TestCandidateConcede(t *testing.T) {
  215. tt := newNetwork(nil, nil, nil)
  216. tt.isolate(0)
  217. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  218. tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
  219. // heal the partition
  220. tt.recover()
  221. data := []byte("force follower")
  222. // send a proposal to 2 to flush out a msgApp to 0
  223. tt.send(pb.Message{From: 2, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: data}}})
  224. a := tt.peers[0].(*raft)
  225. if g := a.state; g != stateFollower {
  226. t.Errorf("state = %s, want %s", g, stateFollower)
  227. }
  228. if g := a.Term; g != 1 {
  229. t.Errorf("term = %d, want %d", g, 1)
  230. }
  231. wantLog := ltoa(&raftLog{ents: []pb.Entry{{}, {Type: Normal, Data: nil, Term: 1, Index: 1}, {Term: 1, Index: 2, Data: data}}, committed: 2})
  232. for i, p := range tt.peers {
  233. if sm, ok := p.(*raft); ok {
  234. l := ltoa(sm.raftLog)
  235. if g := diffu(wantLog, l); g != "" {
  236. t.Errorf("#%d: diff:\n%s", i, g)
  237. }
  238. } else {
  239. t.Logf("#%d: empty log", i)
  240. }
  241. }
  242. }
  243. func TestSingleNodeCandidate(t *testing.T) {
  244. tt := newNetwork(nil)
  245. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  246. sm := tt.peers[0].(*raft)
  247. if sm.state != stateLeader {
  248. t.Errorf("state = %d, want %d", sm.state, stateLeader)
  249. }
  250. }
  251. func TestOldMessages(t *testing.T) {
  252. tt := newNetwork(nil, nil, nil)
  253. // make 0 leader @ term 3
  254. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  255. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  256. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  257. // pretend we're an old leader trying to make progress
  258. tt.send(pb.Message{From: 0, To: 0, Type: msgApp, Term: 1, Entries: []pb.Entry{{Term: 1}}})
  259. l := &raftLog{
  260. ents: []pb.Entry{
  261. {}, {Type: Normal, Data: nil, Term: 1, Index: 1},
  262. {Type: Normal, Data: nil, Term: 2, Index: 2}, {Type: Normal, Data: nil, Term: 3, Index: 3},
  263. },
  264. committed: 3,
  265. }
  266. base := ltoa(l)
  267. for i, p := range tt.peers {
  268. if sm, ok := p.(*raft); ok {
  269. l := ltoa(sm.raftLog)
  270. if g := diffu(base, l); g != "" {
  271. t.Errorf("#%d: diff:\n%s", i, g)
  272. }
  273. } else {
  274. t.Logf("#%d: empty log", i)
  275. }
  276. }
  277. }
  278. // TestOldMessagesReply - optimization - reply with new term.
  279. func TestProposal(t *testing.T) {
  280. tests := []struct {
  281. *network
  282. success bool
  283. }{
  284. {newNetwork(nil, nil, nil), true},
  285. {newNetwork(nil, nil, nopStepper), true},
  286. {newNetwork(nil, nopStepper, nopStepper), false},
  287. {newNetwork(nil, nopStepper, nopStepper, nil), false},
  288. {newNetwork(nil, nopStepper, nopStepper, nil, nil), true},
  289. }
  290. for i, tt := range tests {
  291. send := func(m pb.Message) {
  292. defer func() {
  293. // only recover is we expect it to panic so
  294. // panics we don't expect go up.
  295. if !tt.success {
  296. e := recover()
  297. if e != nil {
  298. t.Logf("#%d: err: %s", i, e)
  299. }
  300. }
  301. }()
  302. tt.send(m)
  303. }
  304. data := []byte("somedata")
  305. // promote 0 the leader
  306. send(pb.Message{From: 0, To: 0, Type: msgHup})
  307. send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Data: data}}})
  308. wantLog := newLog()
  309. if tt.success {
  310. wantLog = &raftLog{ents: []pb.Entry{{}, {Type: Normal, Data: nil, Term: 1, Index: 1}, {Term: 1, Index: 2, Data: data}}, committed: 2}
  311. }
  312. base := ltoa(wantLog)
  313. for i, p := range tt.peers {
  314. if sm, ok := p.(*raft); ok {
  315. l := ltoa(sm.raftLog)
  316. if g := diffu(base, l); g != "" {
  317. t.Errorf("#%d: diff:\n%s", i, g)
  318. }
  319. } else {
  320. t.Logf("#%d: empty log", i)
  321. }
  322. }
  323. sm := tt.network.peers[0].(*raft)
  324. if g := sm.Term; g != 1 {
  325. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  326. }
  327. }
  328. }
  329. func TestProposalByProxy(t *testing.T) {
  330. data := []byte("somedata")
  331. tests := []*network{
  332. newNetwork(nil, nil, nil),
  333. newNetwork(nil, nil, nopStepper),
  334. }
  335. for i, tt := range tests {
  336. // promote 0 the leader
  337. tt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  338. // propose via follower
  339. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  340. wantLog := &raftLog{ents: []pb.Entry{{}, {Type: Normal, Data: nil, Term: 1, Index: 1}, {Term: 1, Data: data, Index: 2}}, committed: 2}
  341. base := ltoa(wantLog)
  342. for i, p := range tt.peers {
  343. if sm, ok := p.(*raft); ok {
  344. l := ltoa(sm.raftLog)
  345. if g := diffu(base, l); g != "" {
  346. t.Errorf("#%d: diff:\n%s", i, g)
  347. }
  348. } else {
  349. t.Logf("#%d: empty log", i)
  350. }
  351. }
  352. sm := tt.peers[0].(*raft)
  353. if g := sm.Term; g != 1 {
  354. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  355. }
  356. }
  357. }
  358. func TestCommit(t *testing.T) {
  359. tests := []struct {
  360. matches []int64
  361. logs []pb.Entry
  362. smTerm int64
  363. w int64
  364. }{
  365. // single
  366. {[]int64{1}, []pb.Entry{{}, {Term: 1}}, 1, 1},
  367. {[]int64{1}, []pb.Entry{{}, {Term: 1}}, 2, 0},
  368. {[]int64{2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 2, 2},
  369. {[]int64{1}, []pb.Entry{{}, {Term: 2}}, 2, 1},
  370. // odd
  371. {[]int64{2, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 1, 1},
  372. {[]int64{2, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  373. {[]int64{2, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 2, 2},
  374. {[]int64{2, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  375. // even
  376. {[]int64{2, 1, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 1, 1},
  377. {[]int64{2, 1, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  378. {[]int64{2, 1, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 1, 1},
  379. {[]int64{2, 1, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  380. {[]int64{2, 1, 2, 2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 2, 2},
  381. {[]int64{2, 1, 2, 2}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  382. }
  383. for i, tt := range tests {
  384. prs := make(map[int64]*progress)
  385. for j := 0; j < len(tt.matches); j++ {
  386. prs[int64(j)] = &progress{tt.matches[j], tt.matches[j] + 1}
  387. }
  388. sm := &raft{raftLog: &raftLog{ents: tt.logs}, prs: prs, State: pb.State{Term: tt.smTerm}}
  389. sm.maybeCommit()
  390. if g := sm.raftLog.committed; g != tt.w {
  391. t.Errorf("#%d: committed = %d, want %d", i, g, tt.w)
  392. }
  393. }
  394. }
  395. // TestHandleMsgApp ensures:
  396. // 1. Reply false if log doesn’t contain an entry at prevLogIndex whose term matches prevLogTerm.
  397. // 2. If an existing entry conflicts with a new one (same index but different terms),
  398. // delete the existing entry and all that follow it; append any new entries not already in the log.
  399. // 3. If leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of last new entry).
  400. func TestHandleMsgApp(t *testing.T) {
  401. tests := []struct {
  402. m pb.Message
  403. wIndex int64
  404. wCommit int64
  405. wAccept bool
  406. }{
  407. // Ensure 1
  408. {pb.Message{Type: msgApp, Term: 2, LogTerm: 3, Index: 2, Commit: 3}, 2, 0, false}, // previous log mismatch
  409. {pb.Message{Type: msgApp, Term: 2, LogTerm: 3, Index: 3, Commit: 3}, 2, 0, false}, // previous log non-exist
  410. // Ensure 2
  411. {pb.Message{Type: msgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 1}, 2, 1, true},
  412. {pb.Message{Type: msgApp, Term: 2, LogTerm: 0, Index: 0, Commit: 1, Entries: []pb.Entry{{Term: 2}}}, 1, 1, true},
  413. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 3, Entries: []pb.Entry{{Term: 2}, {Term: 2}}}, 4, 3, true},
  414. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4, Entries: []pb.Entry{{Term: 2}}}, 3, 3, true},
  415. {pb.Message{Type: msgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 4, Entries: []pb.Entry{{Term: 2}}}, 2, 2, true},
  416. // Ensure 3
  417. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 2}, 2, 2, true},
  418. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4}, 2, 2, true}, // commit upto min(commit, last)
  419. }
  420. for i, tt := range tests {
  421. sm := &raft{
  422. state: stateFollower,
  423. State: pb.State{Term: 2},
  424. raftLog: &raftLog{committed: 0, ents: []pb.Entry{{}, {Term: 1}, {Term: 2}}},
  425. }
  426. sm.handleAppendEntries(tt.m)
  427. if sm.raftLog.lastIndex() != tt.wIndex {
  428. t.Errorf("#%d: lastIndex = %d, want %d", i, sm.raftLog.lastIndex(), tt.wIndex)
  429. }
  430. if sm.raftLog.committed != tt.wCommit {
  431. t.Errorf("#%d: committed = %d, want %d", i, sm.raftLog.committed, tt.wCommit)
  432. }
  433. m := sm.ReadMessages()
  434. if len(m) != 1 {
  435. t.Errorf("#%d: msg = nil, want 1")
  436. }
  437. gaccept := true
  438. if m[0].Index == -1 {
  439. gaccept = false
  440. }
  441. if gaccept != tt.wAccept {
  442. t.Errorf("#%d: accept = %v, want %v", gaccept, tt.wAccept)
  443. }
  444. }
  445. }
  446. func TestRecvMsgVote(t *testing.T) {
  447. tests := []struct {
  448. state stateType
  449. i, term int64
  450. voteFor int64
  451. w int64
  452. }{
  453. {stateFollower, 0, 0, none, -1},
  454. {stateFollower, 0, 1, none, -1},
  455. {stateFollower, 0, 2, none, -1},
  456. {stateFollower, 0, 3, none, 2},
  457. {stateFollower, 1, 0, none, -1},
  458. {stateFollower, 1, 1, none, -1},
  459. {stateFollower, 1, 2, none, -1},
  460. {stateFollower, 1, 3, none, 2},
  461. {stateFollower, 2, 0, none, -1},
  462. {stateFollower, 2, 1, none, -1},
  463. {stateFollower, 2, 2, none, 2},
  464. {stateFollower, 2, 3, none, 2},
  465. {stateFollower, 3, 0, none, -1},
  466. {stateFollower, 3, 1, none, -1},
  467. {stateFollower, 3, 2, none, 2},
  468. {stateFollower, 3, 3, none, 2},
  469. {stateFollower, 3, 2, 1, 2},
  470. {stateFollower, 3, 2, 0, -1},
  471. {stateLeader, 3, 3, 0, -1},
  472. {stateCandidate, 3, 3, 0, -1},
  473. }
  474. for i, tt := range tests {
  475. sm := &raft{
  476. state: tt.state,
  477. State: pb.State{Vote: tt.voteFor},
  478. raftLog: &raftLog{ents: []pb.Entry{{}, {Term: 2}, {Term: 2}}},
  479. }
  480. sm.Step(pb.Message{Type: msgVote, From: 1, Index: tt.i, LogTerm: tt.term})
  481. msgs := sm.ReadMessages()
  482. if g := len(msgs); g != 1 {
  483. t.Errorf("#%d: len(msgs) = %d, want 1", i, g)
  484. continue
  485. }
  486. if g := msgs[0].Index; g != tt.w {
  487. t.Errorf("#%d, m.Index = %d, want %d", i, g, tt.w)
  488. }
  489. }
  490. }
  491. func TestStateTransition(t *testing.T) {
  492. tests := []struct {
  493. from stateType
  494. to stateType
  495. wallow bool
  496. wterm int64
  497. wlead int64
  498. }{
  499. {stateFollower, stateFollower, true, 1, none},
  500. {stateFollower, stateCandidate, true, 1, none},
  501. {stateFollower, stateLeader, false, -1, none},
  502. {stateCandidate, stateFollower, true, 0, none},
  503. {stateCandidate, stateCandidate, true, 1, none},
  504. {stateCandidate, stateLeader, true, 0, 0},
  505. {stateLeader, stateFollower, true, 1, none},
  506. {stateLeader, stateCandidate, false, 1, none},
  507. {stateLeader, stateLeader, true, 0, 0},
  508. }
  509. for i, tt := range tests {
  510. func() {
  511. defer func() {
  512. if r := recover(); r != nil {
  513. if tt.wallow == true {
  514. t.Errorf("%d: allow = %v, want %v", i, false, true)
  515. }
  516. }
  517. }()
  518. sm := newRaft(0, []int64{0}, 0, 0)
  519. sm.state = tt.from
  520. switch tt.to {
  521. case stateFollower:
  522. sm.becomeFollower(tt.wterm, tt.wlead)
  523. case stateCandidate:
  524. sm.becomeCandidate()
  525. case stateLeader:
  526. sm.becomeLeader()
  527. }
  528. if sm.Term != tt.wterm {
  529. t.Errorf("%d: term = %d, want %d", i, sm.Term, tt.wterm)
  530. }
  531. if sm.lead != tt.wlead {
  532. t.Errorf("%d: lead = %d, want %d", i, sm.lead, tt.wlead)
  533. }
  534. }()
  535. }
  536. }
  537. func TestConf(t *testing.T) {
  538. sm := newRaft(0, []int64{0}, 0, 0)
  539. sm.becomeCandidate()
  540. sm.becomeLeader()
  541. sm.Step(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Type: AddNode}}})
  542. if sm.raftLog.lastIndex() != 2 {
  543. t.Errorf("lastindex = %d, want %d", sm.raftLog.lastIndex(), 1)
  544. }
  545. if !sm.configuring {
  546. t.Errorf("pendingConf = %v, want %v", sm.configuring, true)
  547. }
  548. if sm.raftLog.ents[2].Type != AddNode {
  549. t.Errorf("type = %d, want %d", sm.raftLog.ents[1].Type, AddNode)
  550. }
  551. // deny the second configuration change request if there is a pending one
  552. paniced := false
  553. defer func() { recover(); paniced = true }()
  554. sm.Step(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{Type: AddNode}}})
  555. if !paniced {
  556. t.Errorf("expected panic")
  557. }
  558. if sm.raftLog.lastIndex() != 2 {
  559. t.Errorf("lastindex = %d, want %d", sm.raftLog.lastIndex(), 1)
  560. }
  561. }
  562. // Ensures that the new leader sets the pendingConf flag correctly according to
  563. // the uncommitted log entries
  564. func TestConfChangeLeader(t *testing.T) {
  565. tests := []struct {
  566. et int64
  567. wPending bool
  568. }{
  569. {Normal, false},
  570. {AddNode, true},
  571. {RemoveNode, true},
  572. }
  573. for i, tt := range tests {
  574. sm := newRaft(0, []int64{0}, 0, 0)
  575. sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Type: tt.et}}}
  576. sm.becomeCandidate()
  577. sm.becomeLeader()
  578. if sm.configuring != tt.wPending {
  579. t.Errorf("#%d: pendingConf = %v, want %v", i, sm.configuring, tt.wPending)
  580. }
  581. }
  582. }
  583. func TestAllServerStepdown(t *testing.T) {
  584. tests := []struct {
  585. state stateType
  586. wstate stateType
  587. wterm int64
  588. windex int64
  589. }{
  590. {stateFollower, stateFollower, 3, 1},
  591. {stateCandidate, stateFollower, 3, 1},
  592. {stateLeader, stateFollower, 3, 2},
  593. }
  594. tmsgTypes := [...]int64{msgVote, msgApp}
  595. tterm := int64(3)
  596. for i, tt := range tests {
  597. sm := newRaft(0, []int64{0, 1, 2}, 0, 0)
  598. switch tt.state {
  599. case stateFollower:
  600. sm.becomeFollower(1, 0)
  601. case stateCandidate:
  602. sm.becomeCandidate()
  603. case stateLeader:
  604. sm.becomeCandidate()
  605. sm.becomeLeader()
  606. }
  607. for j, msgType := range tmsgTypes {
  608. sm.Step(pb.Message{From: 1, Type: msgType, Term: tterm, LogTerm: tterm})
  609. if sm.state != tt.wstate {
  610. t.Errorf("#%d.%d state = %v , want %v", i, j, sm.state, tt.wstate)
  611. }
  612. if sm.Term != tt.wterm {
  613. t.Errorf("#%d.%d term = %v , want %v", i, j, sm.Term, tt.wterm)
  614. }
  615. if int64(len(sm.raftLog.ents)) != tt.windex {
  616. t.Errorf("#%d.%d index = %v , want %v", i, j, len(sm.raftLog.ents), tt.windex)
  617. }
  618. wlead := int64(1)
  619. if msgType == msgVote {
  620. wlead = none
  621. }
  622. if sm.lead != wlead {
  623. t.Errorf("#%d, sm.lead = %d, want %d", i, sm.lead, none)
  624. }
  625. }
  626. }
  627. }
  628. func TestLeaderAppResp(t *testing.T) {
  629. tests := []struct {
  630. index int64
  631. wmsgNum int
  632. windex int64
  633. wcommitted int64
  634. }{
  635. {-1, 1, 1, 0}, // bad resp; leader does not commit; reply with log entries
  636. {2, 2, 2, 2}, // good resp; leader commits; broadcast with commit index
  637. }
  638. for i, tt := range tests {
  639. // sm term is 1 after it becomes the leader.
  640. // thus the last log term must be 1 to be committed.
  641. sm := newRaft(0, []int64{0, 1, 2}, 0, 0)
  642. sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}}
  643. sm.becomeCandidate()
  644. sm.becomeLeader()
  645. sm.ReadMessages()
  646. sm.Step(pb.Message{From: 1, Type: msgAppResp, Index: tt.index, Term: sm.Term})
  647. msgs := sm.ReadMessages()
  648. if len(msgs) != tt.wmsgNum {
  649. t.Errorf("#%d msgNum = %d, want %d", i, len(msgs), tt.wmsgNum)
  650. }
  651. for j, msg := range msgs {
  652. if msg.Index != tt.windex {
  653. t.Errorf("#%d.%d index = %d, want %d", i, j, msg.Index, tt.windex)
  654. }
  655. if msg.Commit != tt.wcommitted {
  656. t.Errorf("#%d.%d commit = %d, want %d", i, j, msg.Commit, tt.wcommitted)
  657. }
  658. }
  659. }
  660. }
  661. // tests the output of the statemachine when receiving msgBeat
  662. func TestRecvMsgBeat(t *testing.T) {
  663. tests := []struct {
  664. state stateType
  665. wMsg int
  666. }{
  667. {stateLeader, 2},
  668. // candidate and follower should ignore msgBeat
  669. {stateCandidate, 0},
  670. {stateFollower, 0},
  671. }
  672. for i, tt := range tests {
  673. sm := newRaft(0, []int64{0, 1, 2}, 0, 0)
  674. sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}}
  675. sm.Term = 1
  676. sm.state = tt.state
  677. sm.Step(pb.Message{From: 0, To: 0, Type: msgBeat})
  678. msgs := sm.ReadMessages()
  679. if len(msgs) != tt.wMsg {
  680. t.Errorf("%d: len(msgs) = %d, want %d", i, len(msgs), tt.wMsg)
  681. }
  682. for _, m := range msgs {
  683. if m.Type != msgApp {
  684. t.Errorf("%d: msg.type = %v, want %v", m.Type, msgApp)
  685. }
  686. }
  687. }
  688. }
  689. func TestRestore(t *testing.T) {
  690. s := pb.Snapshot{
  691. Index: defaultCompactThreshold + 1,
  692. Term: defaultCompactThreshold + 1,
  693. Nodes: []int64{0, 1, 2},
  694. }
  695. sm := newRaft(0, []int64{0, 1}, 0, 0)
  696. if ok := sm.restore(s); !ok {
  697. t.Fatal("restore fail, want succeed")
  698. }
  699. if sm.raftLog.lastIndex() != s.Index {
  700. t.Errorf("log.lastIndex = %d, want %d", sm.raftLog.lastIndex(), s.Index)
  701. }
  702. if sm.raftLog.term(s.Index) != s.Term {
  703. t.Errorf("log.lastTerm = %d, want %d", sm.raftLog.term(s.Index), s.Term)
  704. }
  705. sg := int64Slice(sm.nodes())
  706. sw := int64Slice(s.Nodes)
  707. sort.Sort(sg)
  708. sort.Sort(sw)
  709. if !reflect.DeepEqual(sg, sw) {
  710. t.Errorf("sm.Nodes = %+v, want %+v", sg, sw)
  711. }
  712. if !reflect.DeepEqual(sm.raftLog.snapshot, s) {
  713. t.Errorf("snapshot = %+v, want %+v", sm.raftLog.snapshot, s)
  714. }
  715. if ok := sm.restore(s); ok {
  716. t.Fatal("restore succeed, want fail")
  717. }
  718. }
  719. func TestProvideSnap(t *testing.T) {
  720. s := pb.Snapshot{
  721. Index: defaultCompactThreshold + 1,
  722. Term: defaultCompactThreshold + 1,
  723. Nodes: []int64{0, 1},
  724. }
  725. sm := newRaft(0, []int64{0}, 0, 0)
  726. // restore the statemachin from a snapshot
  727. // so it has a compacted log and a snapshot
  728. sm.restore(s)
  729. sm.becomeCandidate()
  730. sm.becomeLeader()
  731. sm.Step(pb.Message{From: 0, To: 0, Type: msgBeat})
  732. msgs := sm.ReadMessages()
  733. if len(msgs) != 1 {
  734. t.Errorf("len(msgs) = %d, want 1", len(msgs))
  735. }
  736. m := msgs[0]
  737. if m.Type != msgApp {
  738. t.Errorf("m.Type = %v, want %v", m.Type, msgApp)
  739. }
  740. // force set the next of node 1, so that
  741. // node 1 needs a snapshot
  742. sm.prs[1].next = sm.raftLog.offset
  743. sm.Step(pb.Message{From: 1, To: 0, Type: msgAppResp, Index: -1})
  744. msgs = sm.ReadMessages()
  745. if len(msgs) != 1 {
  746. t.Errorf("len(msgs) = %d, want 1", len(msgs))
  747. }
  748. m = msgs[0]
  749. if m.Type != msgSnap {
  750. t.Errorf("m.Type = %v, want %v", m.Type, msgSnap)
  751. }
  752. }
  753. func TestRestoreFromSnapMsg(t *testing.T) {
  754. s := pb.Snapshot{
  755. Index: defaultCompactThreshold + 1,
  756. Term: defaultCompactThreshold + 1,
  757. Nodes: []int64{0, 1},
  758. }
  759. m := pb.Message{Type: msgSnap, From: 0, Term: 1, Snapshot: s}
  760. sm := newRaft(1, []int64{0, 1}, 0, 0)
  761. sm.Step(m)
  762. if !reflect.DeepEqual(sm.raftLog.snapshot, s) {
  763. t.Errorf("snapshot = %+v, want %+v", sm.raftLog.snapshot, s)
  764. }
  765. }
  766. func TestSlowNodeRestore(t *testing.T) {
  767. nt := newNetwork(nil, nil, nil)
  768. nt.send(pb.Message{From: 0, To: 0, Type: msgHup})
  769. nt.isolate(2)
  770. for j := 0; j < defaultCompactThreshold+1; j++ {
  771. nt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{}}})
  772. }
  773. lead := nt.peers[0].(*raft)
  774. lead.nextEnts()
  775. lead.compact(nil)
  776. nt.recover()
  777. nt.send(pb.Message{From: 0, To: 0, Type: msgBeat})
  778. follower := nt.peers[2].(*raft)
  779. if !reflect.DeepEqual(follower.raftLog.snapshot, lead.raftLog.snapshot) {
  780. t.Errorf("follower.snap = %+v, want %+v", follower.raftLog.snapshot, lead.raftLog.snapshot)
  781. }
  782. committed := follower.raftLog.lastIndex()
  783. nt.send(pb.Message{From: 0, To: 0, Type: msgProp, Entries: []pb.Entry{{}}})
  784. if follower.raftLog.committed != committed+1 {
  785. t.Errorf("follower.comitted = %d, want %d", follower.raftLog.committed, committed+1)
  786. }
  787. }
  788. func ents(terms ...int64) *raft {
  789. ents := []pb.Entry{{}}
  790. for _, term := range terms {
  791. ents = append(ents, pb.Entry{Term: term})
  792. }
  793. sm := &raft{raftLog: &raftLog{ents: ents}}
  794. sm.reset(0)
  795. return sm
  796. }
  797. type network struct {
  798. peers map[int64]Interface
  799. dropm map[connem]float64
  800. ignorem map[int64]bool
  801. }
  802. // newNetwork initializes a network from peers.
  803. // A nil node will be replaced with a new *stateMachine.
  804. // A *stateMachine will get its k, id.
  805. // When using stateMachine, the address list is always [0, n).
  806. func newNetwork(peers ...Interface) *network {
  807. size := len(peers)
  808. defaultPeerAddrs := make([]int64, size)
  809. for i := 0; i < size; i++ {
  810. defaultPeerAddrs[i] = int64(i)
  811. }
  812. npeers := make(map[int64]Interface, size)
  813. for id, p := range peers {
  814. nid := int64(id)
  815. switch v := p.(type) {
  816. case nil:
  817. sm := newRaft(nid, defaultPeerAddrs, 0, 0)
  818. npeers[nid] = sm
  819. case *raft:
  820. v.id = nid
  821. v.prs = make(map[int64]*progress)
  822. for i := 0; i < size; i++ {
  823. v.prs[int64(i)] = &progress{}
  824. }
  825. v.reset(0)
  826. npeers[nid] = v
  827. default:
  828. npeers[nid] = v
  829. }
  830. }
  831. return &network{
  832. peers: npeers,
  833. dropm: make(map[connem]float64),
  834. ignorem: make(map[int64]bool),
  835. }
  836. }
  837. func (nw *network) send(msgs ...pb.Message) {
  838. for len(msgs) > 0 {
  839. m := msgs[0]
  840. p := nw.peers[m.To]
  841. p.Step(m)
  842. msgs = append(msgs[1:], nw.filter(p.ReadMessages())...)
  843. }
  844. }
  845. func (nw *network) drop(from, to int64, perc float64) {
  846. nw.dropm[connem{from, to}] = perc
  847. }
  848. func (nw *network) cut(one, other int64) {
  849. nw.drop(one, other, 1)
  850. nw.drop(other, one, 1)
  851. }
  852. func (nw *network) isolate(id int64) {
  853. for i := 0; i < len(nw.peers); i++ {
  854. nid := int64(i)
  855. if nid != id {
  856. nw.drop(id, nid, 1.0)
  857. nw.drop(nid, id, 1.0)
  858. }
  859. }
  860. }
  861. func (nw *network) ignore(t int64) {
  862. nw.ignorem[t] = true
  863. }
  864. func (nw *network) recover() {
  865. nw.dropm = make(map[connem]float64)
  866. nw.ignorem = make(map[int64]bool)
  867. }
  868. func (nw *network) filter(msgs []pb.Message) []pb.Message {
  869. mm := make([]pb.Message, 0)
  870. for _, m := range msgs {
  871. if nw.ignorem[m.Type] {
  872. continue
  873. }
  874. switch m.Type {
  875. case msgHup:
  876. // hups never go over the network, so don't drop them but panic
  877. panic("unexpected msgHup")
  878. default:
  879. perc := nw.dropm[connem{m.From, m.To}]
  880. if n := rand.Float64(); n < perc {
  881. continue
  882. }
  883. }
  884. mm = append(mm, m)
  885. }
  886. return mm
  887. }
  888. type connem struct {
  889. from, to int64
  890. }
  891. type blackHole struct{}
  892. func (blackHole) Step(pb.Message) error { return nil }
  893. func (blackHole) ReadMessages() []pb.Message { return nil }
  894. var nopStepper = &blackHole{}