kvstore_test.go 21 KB

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