kvstore_test.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811
  1. // Copyright 2015 CoreOS, Inc.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package storage
  15. import (
  16. "crypto/rand"
  17. "encoding/binary"
  18. "math"
  19. "os"
  20. "reflect"
  21. "testing"
  22. "time"
  23. "github.com/coreos/etcd/pkg/testutil"
  24. "github.com/coreos/etcd/storage/backend"
  25. "github.com/coreos/etcd/storage/storagepb"
  26. )
  27. func TestStoreRev(t *testing.T) {
  28. s := newStore(tmpPath)
  29. defer os.Remove(tmpPath)
  30. for i := 0; i < 3; i++ {
  31. s.Put([]byte("foo"), []byte("bar"))
  32. if r := s.Rev(); r != int64(i+1) {
  33. t.Errorf("#%d: rev = %d, want %d", i, r, i+1)
  34. }
  35. }
  36. }
  37. func TestStorePut(t *testing.T) {
  38. tests := []struct {
  39. rev revision
  40. r indexGetResp
  41. wrev revision
  42. wev storagepb.Event
  43. wputrev revision
  44. }{
  45. {
  46. revision{1, 0},
  47. indexGetResp{revision{}, revision{}, 0, ErrRevisionNotFound},
  48. revision{1, 1},
  49. storagepb.Event{
  50. Type: storagepb.PUT,
  51. Kv: &storagepb.KeyValue{
  52. Key: []byte("foo"),
  53. Value: []byte("bar"),
  54. CreateRevision: 2,
  55. ModRevision: 2,
  56. Version: 1,
  57. },
  58. },
  59. revision{2, 0},
  60. },
  61. {
  62. revision{1, 1},
  63. indexGetResp{revision{2, 0}, revision{2, 0}, 1, nil},
  64. revision{1, 2},
  65. storagepb.Event{
  66. Type: storagepb.PUT,
  67. Kv: &storagepb.KeyValue{
  68. Key: []byte("foo"),
  69. Value: []byte("bar"),
  70. CreateRevision: 2,
  71. ModRevision: 2,
  72. Version: 2,
  73. },
  74. },
  75. revision{2, 1},
  76. },
  77. {
  78. revision{2, 0},
  79. indexGetResp{revision{2, 1}, revision{2, 0}, 2, nil},
  80. revision{2, 1},
  81. storagepb.Event{
  82. Type: storagepb.PUT,
  83. Kv: &storagepb.KeyValue{
  84. Key: []byte("foo"),
  85. Value: []byte("bar"),
  86. CreateRevision: 2,
  87. ModRevision: 3,
  88. Version: 3,
  89. },
  90. },
  91. revision{3, 0},
  92. },
  93. }
  94. for i, tt := range tests {
  95. s, b, index := newFakeStore()
  96. s.currentRev = tt.rev
  97. s.tx = b.BatchTx()
  98. index.indexGetRespc <- tt.r
  99. s.put([]byte("foo"), []byte("bar"))
  100. data, err := tt.wev.Marshal()
  101. if err != nil {
  102. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  103. }
  104. wact := []testutil.Action{
  105. {"put", []interface{}{keyBucketName, newTestBytes(tt.wputrev), data}},
  106. }
  107. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  108. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  109. }
  110. wact = []testutil.Action{
  111. {"get", []interface{}{[]byte("foo"), tt.wputrev.main}},
  112. {"put", []interface{}{[]byte("foo"), tt.wputrev}},
  113. }
  114. if g := index.Action(); !reflect.DeepEqual(g, wact) {
  115. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  116. }
  117. if s.currentRev != tt.wrev {
  118. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  119. }
  120. }
  121. }
  122. func TestStoreRange(t *testing.T) {
  123. ev := storagepb.Event{
  124. Type: storagepb.PUT,
  125. Kv: &storagepb.KeyValue{
  126. Key: []byte("foo"),
  127. Value: []byte("bar"),
  128. CreateRevision: 1,
  129. ModRevision: 2,
  130. Version: 1,
  131. },
  132. }
  133. evb, err := ev.Marshal()
  134. if err != nil {
  135. t.Fatal(err)
  136. }
  137. currev := revision{1, 1}
  138. wrev := int64(2)
  139. tests := []struct {
  140. idxr indexRangeResp
  141. r rangeResp
  142. }{
  143. {
  144. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  145. rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
  146. },
  147. {
  148. indexRangeResp{[][]byte{[]byte("foo"), []byte("foo1")}, []revision{{2, 0}, {3, 0}}},
  149. rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
  150. },
  151. }
  152. for i, tt := range tests {
  153. s, b, index := newFakeStore()
  154. s.currentRev = currev
  155. s.tx = b.BatchTx()
  156. b.tx.rangeRespc <- tt.r
  157. index.indexRangeRespc <- tt.idxr
  158. kvs, rev, err := s.rangeKeys([]byte("foo"), []byte("goo"), 1, 0)
  159. if err != nil {
  160. t.Errorf("#%d: err = %v, want nil", i, err)
  161. }
  162. if w := []storagepb.KeyValue{*ev.Kv}; !reflect.DeepEqual(kvs, w) {
  163. t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, w)
  164. }
  165. if rev != wrev {
  166. t.Errorf("#%d: rev = %d, want %d", i, rev, wrev)
  167. }
  168. wact := []testutil.Action{
  169. {"range", []interface{}{keyBucketName, newTestBytes(tt.idxr.revs[0]), []byte(nil), int64(0)}},
  170. }
  171. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  172. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  173. }
  174. wact = []testutil.Action{
  175. {"range", []interface{}{[]byte("foo"), []byte("goo"), wrev}},
  176. }
  177. if g := index.Action(); !reflect.DeepEqual(g, wact) {
  178. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  179. }
  180. if s.currentRev != currev {
  181. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, currev)
  182. }
  183. }
  184. }
  185. func TestStoreDeleteRange(t *testing.T) {
  186. tests := []struct {
  187. rev revision
  188. r indexRangeResp
  189. wrev revision
  190. wrrev int64
  191. wdelrev revision
  192. }{
  193. {
  194. revision{2, 0},
  195. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  196. revision{2, 1},
  197. 2,
  198. revision{3, 0},
  199. },
  200. {
  201. revision{2, 1},
  202. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  203. revision{2, 2},
  204. 3,
  205. revision{3, 1},
  206. },
  207. }
  208. for i, tt := range tests {
  209. s, b, index := newFakeStore()
  210. s.currentRev = tt.rev
  211. s.tx = b.BatchTx()
  212. index.indexRangeRespc <- tt.r
  213. n := s.deleteRange([]byte("foo"), []byte("goo"))
  214. if n != 1 {
  215. t.Errorf("#%d: n = %d, want 1", i, n)
  216. }
  217. data, err := (&storagepb.Event{
  218. Type: storagepb.DELETE,
  219. Kv: &storagepb.KeyValue{
  220. Key: []byte("foo"),
  221. },
  222. }).Marshal()
  223. if err != nil {
  224. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  225. }
  226. wact := []testutil.Action{
  227. {"put", []interface{}{keyBucketName, newTestBytes(tt.wdelrev), data}},
  228. }
  229. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  230. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  231. }
  232. wact = []testutil.Action{
  233. {"range", []interface{}{[]byte("foo"), []byte("goo"), tt.wrrev}},
  234. {"tombstone", []interface{}{[]byte("foo"), tt.wdelrev}},
  235. }
  236. if g := index.Action(); !reflect.DeepEqual(g, wact) {
  237. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  238. }
  239. if s.currentRev != tt.wrev {
  240. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  241. }
  242. }
  243. }
  244. func TestStoreRangeEvents(t *testing.T) {
  245. ev := storagepb.Event{
  246. Type: storagepb.PUT,
  247. Kv: &storagepb.KeyValue{
  248. Key: []byte("foo"),
  249. Value: []byte("bar"),
  250. CreateRevision: 1,
  251. ModRevision: 2,
  252. Version: 1,
  253. },
  254. }
  255. evb, err := ev.Marshal()
  256. if err != nil {
  257. t.Fatal(err)
  258. }
  259. currev := revision{2, 0}
  260. tests := []struct {
  261. idxr indexRangeEventsResp
  262. r rangeResp
  263. }{
  264. {
  265. indexRangeEventsResp{[]revision{{2, 0}}},
  266. rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
  267. },
  268. {
  269. indexRangeEventsResp{[]revision{{2, 0}, {3, 0}}},
  270. rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
  271. },
  272. }
  273. for i, tt := range tests {
  274. s, b, index := newFakeStore()
  275. s.currentRev = currev
  276. index.indexRangeEventsRespc <- tt.idxr
  277. b.tx.rangeRespc <- tt.r
  278. evs, _, err := s.RangeEvents([]byte("foo"), []byte("goo"), 1, 1, 4)
  279. if err != nil {
  280. t.Errorf("#%d: err = %v, want nil", i, err)
  281. }
  282. if w := []storagepb.Event{ev}; !reflect.DeepEqual(evs, w) {
  283. t.Errorf("#%d: evs = %+v, want %+v", i, evs, w)
  284. }
  285. wact := []testutil.Action{
  286. {"rangeEvents", []interface{}{[]byte("foo"), []byte("goo"), int64(1)}},
  287. }
  288. if g := index.Action(); !reflect.DeepEqual(g, wact) {
  289. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  290. }
  291. wact = []testutil.Action{
  292. {"range", []interface{}{keyBucketName, newTestBytes(tt.idxr.revs[0]), []byte(nil), int64(0)}},
  293. }
  294. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  295. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  296. }
  297. if s.currentRev != currev {
  298. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, currev)
  299. }
  300. }
  301. }
  302. func TestStoreCompact(t *testing.T) {
  303. s, b, index := newFakeStore()
  304. s.currentRev = revision{3, 0}
  305. index.indexCompactRespc <- map[revision]struct{}{revision{1, 0}: {}}
  306. b.tx.rangeRespc <- rangeResp{[][]byte{newTestBytes(revision{1, 0}), newTestBytes(revision{2, 0})}, nil}
  307. s.Compact(3)
  308. s.wg.Wait()
  309. if s.compactMainRev != 3 {
  310. t.Errorf("compact main rev = %d, want 3", s.compactMainRev)
  311. }
  312. end := make([]byte, 8)
  313. binary.BigEndian.PutUint64(end, uint64(4))
  314. wact := []testutil.Action{
  315. {"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestBytes(revision{3, 0})}},
  316. {"range", []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
  317. {"delete", []interface{}{keyBucketName, newTestBytes(revision{2, 0})}},
  318. {"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestBytes(revision{3, 0})}},
  319. }
  320. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  321. t.Errorf("tx actions = %+v, want %+v", g, wact)
  322. }
  323. wact = []testutil.Action{
  324. {"compact", []interface{}{int64(3)}},
  325. }
  326. if g := index.Action(); !reflect.DeepEqual(g, wact) {
  327. t.Errorf("index action = %+v, want %+v", g, wact)
  328. }
  329. }
  330. func TestStoreRestore(t *testing.T) {
  331. s, b, index := newFakeStore()
  332. putev := storagepb.Event{
  333. Type: storagepb.PUT,
  334. Kv: &storagepb.KeyValue{
  335. Key: []byte("foo"),
  336. Value: []byte("bar"),
  337. CreateRevision: 3,
  338. ModRevision: 3,
  339. Version: 1,
  340. },
  341. }
  342. putevb, err := putev.Marshal()
  343. if err != nil {
  344. t.Fatal(err)
  345. }
  346. delev := storagepb.Event{
  347. Type: storagepb.DELETE,
  348. Kv: &storagepb.KeyValue{
  349. Key: []byte("foo"),
  350. },
  351. }
  352. delevb, err := delev.Marshal()
  353. if err != nil {
  354. t.Fatal(err)
  355. }
  356. b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestBytes(revision{2, 0})}}
  357. b.tx.rangeRespc <- rangeResp{[][]byte{newTestBytes(revision{3, 0}), newTestBytes(revision{4, 0})}, [][]byte{putevb, delevb}}
  358. b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestBytes(revision{2, 0})}}
  359. s.Restore()
  360. if s.compactMainRev != 2 {
  361. t.Errorf("compact rev = %d, want 4", s.compactMainRev)
  362. }
  363. wrev := revision{4, 0}
  364. if !reflect.DeepEqual(s.currentRev, wrev) {
  365. t.Errorf("current rev = %v, want %v", s.currentRev, wrev)
  366. }
  367. wact := []testutil.Action{
  368. {"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
  369. {"range", []interface{}{keyBucketName, newTestBytes(revision{}), newTestBytes(revision{math.MaxInt64, math.MaxInt64}), int64(0)}},
  370. {"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
  371. }
  372. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  373. t.Errorf("tx actions = %+v, want %+v", g, wact)
  374. }
  375. wact = []testutil.Action{
  376. {"restore", []interface{}{[]byte("foo"), revision{3, 0}, revision{3, 0}, int64(1)}},
  377. {"tombstone", []interface{}{[]byte("foo"), revision{4, 0}}},
  378. }
  379. if g := index.Action(); !reflect.DeepEqual(g, wact) {
  380. t.Errorf("index action = %+v, want %+v", g, wact)
  381. }
  382. }
  383. // tests end parameter works well
  384. func TestStoreRangeEventsEnd(t *testing.T) {
  385. s := newStore(tmpPath)
  386. defer cleanup(s, tmpPath)
  387. s.Put([]byte("foo"), []byte("bar"))
  388. s.Put([]byte("foo1"), []byte("bar1"))
  389. s.Put([]byte("foo2"), []byte("bar2"))
  390. evs := []storagepb.Event{
  391. {
  392. Type: storagepb.PUT,
  393. Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
  394. },
  395. {
  396. Type: storagepb.PUT,
  397. Kv: &storagepb.KeyValue{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
  398. },
  399. {
  400. Type: storagepb.PUT,
  401. Kv: &storagepb.KeyValue{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
  402. },
  403. }
  404. tests := []struct {
  405. key, end []byte
  406. wevs []storagepb.Event
  407. }{
  408. // get no keys
  409. {
  410. []byte("doo"), []byte("foo"),
  411. nil,
  412. },
  413. // get no keys when key == end
  414. {
  415. []byte("foo"), []byte("foo"),
  416. nil,
  417. },
  418. // get no keys when ranging single key
  419. {
  420. []byte("doo"), nil,
  421. nil,
  422. },
  423. // get all keys
  424. {
  425. []byte("foo"), []byte("foo3"),
  426. evs,
  427. },
  428. // get partial keys
  429. {
  430. []byte("foo"), []byte("foo1"),
  431. evs[:1],
  432. },
  433. // get single key
  434. {
  435. []byte("foo"), nil,
  436. evs[:1],
  437. },
  438. }
  439. for i, tt := range tests {
  440. evs, rev, err := s.RangeEvents(tt.key, tt.end, 0, 1, 100)
  441. if err != nil {
  442. t.Fatal(err)
  443. }
  444. if rev != 4 {
  445. t.Errorf("#%d: rev = %d, want %d", i, rev, 4)
  446. }
  447. if !reflect.DeepEqual(evs, tt.wevs) {
  448. t.Errorf("#%d: evs = %+v, want %+v", i, evs, tt.wevs)
  449. }
  450. }
  451. }
  452. func TestStoreRangeEventsRev(t *testing.T) {
  453. s := newStore(tmpPath)
  454. defer cleanup(s, tmpPath)
  455. s.Put([]byte("foo"), []byte("bar"))
  456. s.DeleteRange([]byte("foo"), nil)
  457. s.Put([]byte("foo"), []byte("bar"))
  458. s.Put([]byte("unrelated"), []byte("unrelated"))
  459. evs := []storagepb.Event{
  460. {
  461. Type: storagepb.PUT,
  462. Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
  463. },
  464. {
  465. Type: storagepb.DELETE,
  466. Kv: &storagepb.KeyValue{Key: []byte("foo")},
  467. },
  468. {
  469. Type: storagepb.PUT,
  470. Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
  471. },
  472. }
  473. tests := []struct {
  474. start int64
  475. end int64
  476. wevs []storagepb.Event
  477. wnext int64
  478. }{
  479. {1, 1, nil, 1},
  480. {1, 2, evs[:1], 2},
  481. {1, 3, evs[:2], 3},
  482. {1, 4, evs, 5},
  483. {1, 5, evs, 5},
  484. {1, 10, evs, 5},
  485. {3, 4, evs[2:], 5},
  486. {0, 10, evs, 5},
  487. {1, 0, evs, 5},
  488. {0, 0, evs, 5},
  489. }
  490. for i, tt := range tests {
  491. evs, next, err := s.RangeEvents([]byte("foo"), nil, 0, tt.start, tt.end)
  492. if err != nil {
  493. t.Fatal(err)
  494. }
  495. if !reflect.DeepEqual(evs, tt.wevs) {
  496. t.Errorf("#%d: evs = %+v, want %+v", i, evs, tt.wevs)
  497. }
  498. if next != tt.wnext {
  499. t.Errorf("#%d: next = %d, want %d", i, next, tt.wnext)
  500. }
  501. }
  502. }
  503. func TestStoreRangeEventsBad(t *testing.T) {
  504. s := newStore(tmpPath)
  505. defer cleanup(s, tmpPath)
  506. s.Put([]byte("foo"), []byte("bar"))
  507. s.Put([]byte("foo"), []byte("bar1"))
  508. s.Put([]byte("foo"), []byte("bar2"))
  509. if err := s.Compact(3); err != nil {
  510. t.Fatalf("compact error (%v)", err)
  511. }
  512. tests := []struct {
  513. rev int64
  514. werr error
  515. }{
  516. {1, ErrCompacted},
  517. {2, ErrCompacted},
  518. {3, ErrCompacted},
  519. {4, ErrFutureRev},
  520. {10, ErrFutureRev},
  521. }
  522. for i, tt := range tests {
  523. _, _, err := s.RangeEvents([]byte("foo"), nil, 0, tt.rev, 100)
  524. if err != tt.werr {
  525. t.Errorf("#%d: error = %v, want %v", i, err, tt.werr)
  526. }
  527. }
  528. }
  529. func TestStoreRangeEventsLimit(t *testing.T) {
  530. s := newStore(tmpPath)
  531. defer cleanup(s, tmpPath)
  532. s.Put([]byte("foo"), []byte("bar"))
  533. s.DeleteRange([]byte("foo"), nil)
  534. s.Put([]byte("foo"), []byte("bar"))
  535. evs := []storagepb.Event{
  536. {
  537. Type: storagepb.PUT,
  538. Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
  539. },
  540. {
  541. Type: storagepb.DELETE,
  542. Kv: &storagepb.KeyValue{Key: []byte("foo")},
  543. },
  544. {
  545. Type: storagepb.PUT,
  546. Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
  547. },
  548. }
  549. tests := []struct {
  550. limit int64
  551. wevs []storagepb.Event
  552. }{
  553. // no limit
  554. {-1, evs},
  555. // no limit
  556. {0, evs},
  557. {1, evs[:1]},
  558. {2, evs[:2]},
  559. {3, evs},
  560. {100, evs},
  561. }
  562. for i, tt := range tests {
  563. evs, _, err := s.RangeEvents([]byte("foo"), nil, tt.limit, 1, 100)
  564. if err != nil {
  565. t.Fatalf("#%d: range error (%v)", i, err)
  566. }
  567. if !reflect.DeepEqual(evs, tt.wevs) {
  568. t.Errorf("#%d: evs = %+v, want %+v", i, evs, tt.wevs)
  569. }
  570. }
  571. }
  572. func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
  573. s0 := newStore(tmpPath)
  574. defer os.Remove(tmpPath)
  575. s0.Put([]byte("foo"), []byte("bar"))
  576. s0.Put([]byte("foo"), []byte("bar1"))
  577. s0.Put([]byte("foo"), []byte("bar2"))
  578. // write scheduled compaction, but not do compaction
  579. rbytes := newRevBytes()
  580. revToBytes(revision{main: 2}, rbytes)
  581. tx := s0.b.BatchTx()
  582. tx.Lock()
  583. tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
  584. tx.Unlock()
  585. s0.Close()
  586. s1 := newStore(tmpPath)
  587. s1.Restore()
  588. // wait for scheduled compaction to be finished
  589. time.Sleep(100 * time.Millisecond)
  590. if _, _, err := s1.Range([]byte("foo"), nil, 0, 2); err != ErrCompacted {
  591. t.Errorf("range on compacted rev error = %v, want %v", err, ErrCompacted)
  592. }
  593. // check the key in backend is deleted
  594. revbytes := newRevBytes()
  595. // TODO: compact should delete main=2 key too
  596. revToBytes(revision{main: 1}, revbytes)
  597. tx = s1.b.BatchTx()
  598. tx.Lock()
  599. ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
  600. if len(ks) != 0 {
  601. t.Errorf("key for rev %+v still exists, want deleted", bytesToRev(revbytes))
  602. }
  603. tx.Unlock()
  604. }
  605. func TestTxnBlockBackendForceCommit(t *testing.T) {
  606. s := newStore(tmpPath)
  607. defer os.Remove(tmpPath)
  608. id := s.TxnBegin()
  609. done := make(chan struct{})
  610. go func() {
  611. s.b.ForceCommit()
  612. done <- struct{}{}
  613. }()
  614. select {
  615. case <-done:
  616. t.Fatalf("failed to block ForceCommit")
  617. case <-time.After(100 * time.Millisecond):
  618. }
  619. s.TxnEnd(id)
  620. select {
  621. case <-done:
  622. case <-time.After(100 * time.Millisecond):
  623. t.Fatalf("failed to execute ForceCommit")
  624. }
  625. }
  626. func BenchmarkStorePut(b *testing.B) {
  627. s := newStore(tmpPath)
  628. defer os.Remove(tmpPath)
  629. // prepare keys
  630. keys := make([][]byte, b.N)
  631. for i := 0; i < b.N; i++ {
  632. keys[i] = make([]byte, 64)
  633. rand.Read(keys[i])
  634. }
  635. b.ResetTimer()
  636. for i := 0; i < b.N; i++ {
  637. s.Put(keys[i], []byte("foo"))
  638. }
  639. }
  640. func newTestBytes(rev revision) []byte {
  641. bytes := newRevBytes()
  642. revToBytes(rev, bytes)
  643. return bytes
  644. }
  645. func newFakeStore() (*store, *fakeBackend, *fakeIndex) {
  646. b := &fakeBackend{&fakeBatchTx{rangeRespc: make(chan rangeResp, 5)}}
  647. index := &fakeIndex{
  648. indexGetRespc: make(chan indexGetResp, 1),
  649. indexRangeRespc: make(chan indexRangeResp, 1),
  650. indexRangeEventsRespc: make(chan indexRangeEventsResp, 1),
  651. indexCompactRespc: make(chan map[revision]struct{}, 1),
  652. }
  653. return &store{
  654. b: b,
  655. kvindex: index,
  656. currentRev: revision{},
  657. compactMainRev: -1,
  658. }, b, index
  659. }
  660. type rangeResp struct {
  661. keys [][]byte
  662. vals [][]byte
  663. }
  664. type fakeBatchTx struct {
  665. testutil.Recorder
  666. rangeRespc chan rangeResp
  667. }
  668. func (b *fakeBatchTx) Lock() {}
  669. func (b *fakeBatchTx) Unlock() {}
  670. func (b *fakeBatchTx) UnsafeCreateBucket(name []byte) {}
  671. func (b *fakeBatchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
  672. b.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{bucketName, key, value}})
  673. }
  674. func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
  675. b.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{bucketName, key, endKey, limit}})
  676. r := <-b.rangeRespc
  677. return r.keys, r.vals
  678. }
  679. func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
  680. b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
  681. }
  682. func (b *fakeBatchTx) Commit() {}
  683. func (b *fakeBatchTx) CommitAndStop() {}
  684. type fakeBackend struct {
  685. tx *fakeBatchTx
  686. }
  687. func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
  688. func (b *fakeBackend) Hash() (uint32, error) { return 0, nil }
  689. func (b *fakeBackend) Size() int64 { return 0 }
  690. func (b *fakeBackend) Snapshot() backend.Snapshot { return nil }
  691. func (b *fakeBackend) ForceCommit() {}
  692. func (b *fakeBackend) Close() error { return nil }
  693. type indexGetResp struct {
  694. rev revision
  695. created revision
  696. ver int64
  697. err error
  698. }
  699. type indexRangeResp struct {
  700. keys [][]byte
  701. revs []revision
  702. }
  703. type indexRangeEventsResp struct {
  704. revs []revision
  705. }
  706. type fakeIndex struct {
  707. testutil.Recorder
  708. indexGetRespc chan indexGetResp
  709. indexRangeRespc chan indexRangeResp
  710. indexRangeEventsRespc chan indexRangeEventsResp
  711. indexCompactRespc chan map[revision]struct{}
  712. }
  713. func (i *fakeIndex) Get(key []byte, atRev int64) (rev, created revision, ver int64, err error) {
  714. i.Recorder.Record(testutil.Action{Name: "get", Params: []interface{}{key, atRev}})
  715. r := <-i.indexGetRespc
  716. return r.rev, r.created, r.ver, r.err
  717. }
  718. func (i *fakeIndex) Range(key, end []byte, atRev int64) ([][]byte, []revision) {
  719. i.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{key, end, atRev}})
  720. r := <-i.indexRangeRespc
  721. return r.keys, r.revs
  722. }
  723. func (i *fakeIndex) Put(key []byte, rev revision) {
  724. i.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{key, rev}})
  725. }
  726. func (i *fakeIndex) Restore(key []byte, created, modified revision, ver int64) {
  727. i.Recorder.Record(testutil.Action{Name: "restore", Params: []interface{}{key, created, modified, ver}})
  728. }
  729. func (i *fakeIndex) Tombstone(key []byte, rev revision) error {
  730. i.Recorder.Record(testutil.Action{Name: "tombstone", Params: []interface{}{key, rev}})
  731. return nil
  732. }
  733. func (i *fakeIndex) RangeEvents(key, end []byte, rev int64) []revision {
  734. i.Recorder.Record(testutil.Action{Name: "rangeEvents", Params: []interface{}{key, end, rev}})
  735. r := <-i.indexRangeEventsRespc
  736. return r.revs
  737. }
  738. func (i *fakeIndex) Compact(rev int64) map[revision]struct{} {
  739. i.Recorder.Record(testutil.Action{Name: "compact", Params: []interface{}{rev}})
  740. return <-i.indexCompactRespc
  741. }
  742. func (i *fakeIndex) Equal(b index) bool { return false }