kvstore_test.go 20 KB

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