raft_test.go 23 KB

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