raft_test.go 25 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045
  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[int64(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[int64]*index)
  375. for j := 0; j < len(tt.matches); j++ {
  376. ins[int64(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 int64
  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 int64
  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, []int64{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, []int64{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, []int64{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, []int64{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, []int64{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, []int64{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, []int64{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. sw := int64Slice(w)
  647. sg := int64Slice(s.Nodes)
  648. sort.Sort(sw)
  649. sort.Sort(sg)
  650. if !reflect.DeepEqual(sg, sw) {
  651. t.Errorf("#%d: snap.Nodes = %+v, want %+v", i, sg, sw)
  652. }
  653. }
  654. }
  655. }
  656. func TestRestore(t *testing.T) {
  657. s := Snapshot{
  658. Index: defaultCompactThreshold + 1,
  659. Term: defaultCompactThreshold + 1,
  660. Nodes: []int64{0, 1, 2},
  661. }
  662. tests := []struct {
  663. snapshoter Snapshoter
  664. wallow bool
  665. }{
  666. {nil, false},
  667. {new(logSnapshoter), true},
  668. }
  669. for i, tt := range tests {
  670. func() {
  671. defer func() {
  672. if r := recover(); r != nil {
  673. if tt.wallow == true {
  674. t.Errorf("%d: allow = %v, want %v", i, false, true)
  675. }
  676. }
  677. }()
  678. sm := newStateMachine(0, []int64{0, 1})
  679. sm.setSnapshoter(tt.snapshoter)
  680. sm.restore(s)
  681. if sm.log.lastIndex() != s.Index {
  682. t.Errorf("#%d: log.lastIndex = %d, want %d", i, sm.log.lastIndex(), s.Index)
  683. }
  684. if sm.log.term(s.Index) != s.Term {
  685. t.Errorf("#%d: log.lastTerm = %d, want %d", i, sm.log.term(s.Index), s.Term)
  686. }
  687. sg := int64Slice(sm.nodes())
  688. sw := int64Slice(s.Nodes)
  689. sort.Sort(sg)
  690. sort.Sort(sw)
  691. if !reflect.DeepEqual(sg, sw) {
  692. t.Errorf("#%d: sm.Nodes = %+v, want %+v", i, sg, sw)
  693. }
  694. if !reflect.DeepEqual(sm.snapshoter.GetSnap(), s) {
  695. t.Errorf("%d: snapshoter.getSnap = %+v, want %+v", sm.snapshoter.GetSnap(), s)
  696. }
  697. }()
  698. }
  699. }
  700. func TestProvideSnap(t *testing.T) {
  701. s := Snapshot{
  702. Index: defaultCompactThreshold + 1,
  703. Term: defaultCompactThreshold + 1,
  704. Nodes: []int64{0, 1},
  705. }
  706. sm := newStateMachine(0, []int64{0})
  707. sm.setSnapshoter(new(logSnapshoter))
  708. // restore the statemachin from a snapshot
  709. // so it has a compacted log and a snapshot
  710. sm.restore(s)
  711. sm.becomeCandidate()
  712. sm.becomeLeader()
  713. sm.Step(Message{Type: msgBeat})
  714. msgs := sm.Msgs()
  715. if len(msgs) != 1 {
  716. t.Errorf("len(msgs) = %d, want 1", len(msgs))
  717. }
  718. m := msgs[0]
  719. if m.Type != msgApp {
  720. t.Errorf("m.Type = %v, want %v", m.Type, msgApp)
  721. }
  722. // force set the next of node 1, so that
  723. // node 1 needs a snapshot
  724. sm.ins[1].next = sm.log.offset
  725. sm.Step(Message{Type: msgBeat})
  726. msgs = sm.Msgs()
  727. if len(msgs) != 1 {
  728. t.Errorf("len(msgs) = %d, want 1", len(msgs))
  729. }
  730. m = msgs[0]
  731. if m.Type != msgSnap {
  732. t.Errorf("m.Type = %v, want %v", m.Type, msgSnap)
  733. }
  734. }
  735. func TestRestoreFromSnapMsg(t *testing.T) {
  736. s := Snapshot{
  737. Index: defaultCompactThreshold + 1,
  738. Term: defaultCompactThreshold + 1,
  739. Nodes: []int64{0, 1},
  740. }
  741. m := Message{Type: msgSnap, From: 0, Term: 1, Snapshot: s}
  742. sm := newStateMachine(1, []int64{0, 1})
  743. sm.setSnapshoter(new(logSnapshoter))
  744. sm.Step(m)
  745. if !reflect.DeepEqual(sm.snapshoter.GetSnap(), s) {
  746. t.Errorf("snapshot = %+v, want %+v", sm.snapshoter.GetSnap(), s)
  747. }
  748. }
  749. func TestSlowNodeRestore(t *testing.T) {
  750. nt := newNetwork(nil, nil, nil)
  751. nt.send(Message{To: 0, Type: msgHup})
  752. nt.isolate(2)
  753. for j := 0; j < defaultCompactThreshold+1; j++ {
  754. nt.send(Message{To: 0, Type: msgProp, Entries: []Entry{{}}})
  755. }
  756. lead := nt.peers[0].(*stateMachine)
  757. lead.nextEnts()
  758. if !lead.maybeCompact() {
  759. t.Errorf("compacted = false, want true")
  760. }
  761. nt.recover()
  762. nt.send(Message{To: 0, Type: msgBeat})
  763. follower := nt.peers[2].(*stateMachine)
  764. if !reflect.DeepEqual(follower.snapshoter.GetSnap(), lead.snapshoter.GetSnap()) {
  765. t.Errorf("follower.snap = %+v, want %+v", follower.snapshoter.GetSnap(), lead.snapshoter.GetSnap())
  766. }
  767. committed := follower.log.lastIndex()
  768. nt.send(Message{To: 0, Type: msgProp, Entries: []Entry{{}}})
  769. if follower.log.committed != committed+1 {
  770. t.Errorf("follower.comitted = %d, want %d", follower.log.committed, committed+1)
  771. }
  772. }
  773. func ents(terms ...int) *stateMachine {
  774. ents := []Entry{{}}
  775. for _, term := range terms {
  776. ents = append(ents, Entry{Term: term})
  777. }
  778. sm := &stateMachine{log: &log{ents: ents}}
  779. sm.reset(0)
  780. return sm
  781. }
  782. type network struct {
  783. peers map[int64]Interface
  784. dropm map[connem]float64
  785. ignorem map[messageType]bool
  786. }
  787. // newNetwork initializes a network from peers.
  788. // A nil node will be replaced with a new *stateMachine.
  789. // A *stateMachine will get its k, id.
  790. // When using stateMachine, the address list is always [0, n).
  791. func newNetwork(peers ...Interface) *network {
  792. size := len(peers)
  793. defaultPeerAddrs := make([]int64, size)
  794. for i := 0; i < size; i++ {
  795. defaultPeerAddrs[i] = int64(i)
  796. }
  797. npeers := make(map[int64]Interface, size)
  798. for id, p := range peers {
  799. nid := int64(id)
  800. switch v := p.(type) {
  801. case nil:
  802. sm := newStateMachine(nid, defaultPeerAddrs)
  803. sm.setSnapshoter(new(logSnapshoter))
  804. npeers[nid] = sm
  805. case *stateMachine:
  806. v.id = nid
  807. v.ins = make(map[int64]*index)
  808. for i := 0; i < size; i++ {
  809. v.ins[int64(i)] = &index{}
  810. }
  811. v.reset(0)
  812. npeers[nid] = v
  813. case *Node:
  814. npeers[v.sm.id] = v
  815. default:
  816. npeers[nid] = v
  817. }
  818. }
  819. return &network{
  820. peers: npeers,
  821. dropm: make(map[connem]float64),
  822. ignorem: make(map[messageType]bool),
  823. }
  824. }
  825. func (nw *network) send(msgs ...Message) {
  826. for len(msgs) > 0 {
  827. m := msgs[0]
  828. p := nw.peers[m.To]
  829. p.Step(m)
  830. msgs = append(msgs[1:], nw.filter(p.Msgs())...)
  831. }
  832. }
  833. func (nw *network) drop(from, to int64, perc float64) {
  834. nw.dropm[connem{from, to}] = perc
  835. }
  836. func (nw *network) cut(one, other int64) {
  837. nw.drop(one, other, 1)
  838. nw.drop(other, one, 1)
  839. }
  840. func (nw *network) isolate(id int64) {
  841. for i := 0; i < len(nw.peers); i++ {
  842. nid := int64(i)
  843. if nid != id {
  844. nw.drop(id, nid, 1.0)
  845. nw.drop(nid, id, 1.0)
  846. }
  847. }
  848. }
  849. func (nw *network) ignore(t messageType) {
  850. nw.ignorem[t] = true
  851. }
  852. func (nw *network) recover() {
  853. nw.dropm = make(map[connem]float64)
  854. nw.ignorem = make(map[messageType]bool)
  855. }
  856. func (nw *network) filter(msgs []Message) []Message {
  857. mm := make([]Message, 0)
  858. for _, m := range msgs {
  859. if nw.ignorem[m.Type] {
  860. continue
  861. }
  862. switch m.Type {
  863. case msgHup:
  864. // hups never go over the network, so don't drop them but panic
  865. panic("unexpected msgHup")
  866. default:
  867. perc := nw.dropm[connem{m.From, m.To}]
  868. if n := rand.Float64(); n < perc {
  869. continue
  870. }
  871. }
  872. mm = append(mm, m)
  873. }
  874. return mm
  875. }
  876. type connem struct {
  877. from, to int64
  878. }
  879. type blackHole struct{}
  880. func (blackHole) Step(Message) bool { return true }
  881. func (blackHole) Msgs() []Message { return nil }
  882. var nopStepper = &blackHole{}
  883. type logSnapshoter struct {
  884. snapshot Snapshot
  885. }
  886. func (s *logSnapshoter) Snap(index, term int, nodes []int64) {
  887. s.snapshot = Snapshot{
  888. Index: index,
  889. Term: term,
  890. Nodes: nodes,
  891. Data: []byte(fmt.Sprintf("%d:%d", term, index)),
  892. }
  893. }
  894. func (s *logSnapshoter) Restore(ss Snapshot) {
  895. s.snapshot = ss
  896. }
  897. func (s *logSnapshoter) GetSnap() Snapshot {
  898. return s.snapshot
  899. }
  900. // int64Slice implements sort interface
  901. type int64Slice []int64
  902. func (p int64Slice) Len() int { return len(p) }
  903. func (p int64Slice) Less(i, j int) bool { return p[i] < p[j] }
  904. func (p int64Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }