kvstore_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856
  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. mrand "math/rand"
  20. "os"
  21. "reflect"
  22. "testing"
  23. "time"
  24. "github.com/coreos/etcd/pkg/testutil"
  25. "github.com/coreos/etcd/storage/backend"
  26. "github.com/coreos/etcd/storage/storagepb"
  27. )
  28. func TestStoreRev(t *testing.T) {
  29. s := newStore(tmpPath)
  30. defer os.Remove(tmpPath)
  31. for i := 0; i < 3; i++ {
  32. s.Put([]byte("foo"), []byte("bar"))
  33. if r := s.Rev(); r != int64(i+1) {
  34. t.Errorf("#%d: rev = %d, want %d", i, r, i+1)
  35. }
  36. }
  37. }
  38. func TestStorePut(t *testing.T) {
  39. tests := []struct {
  40. rev revision
  41. r indexGetResp
  42. wrev revision
  43. wkey []byte
  44. wkv storagepb.KeyValue
  45. wputrev revision
  46. }{
  47. {
  48. revision{1, 0},
  49. indexGetResp{revision{}, revision{}, 0, ErrRevisionNotFound},
  50. revision{1, 1},
  51. newTestKeyBytes(revision{2, 0}, false),
  52. storagepb.KeyValue{
  53. Key: []byte("foo"),
  54. Value: []byte("bar"),
  55. CreateRevision: 2,
  56. ModRevision: 2,
  57. Version: 1,
  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. newTestKeyBytes(revision{2, 1}, false),
  66. storagepb.KeyValue{
  67. Key: []byte("foo"),
  68. Value: []byte("bar"),
  69. CreateRevision: 2,
  70. ModRevision: 2,
  71. Version: 2,
  72. },
  73. revision{2, 1},
  74. },
  75. {
  76. revision{2, 0},
  77. indexGetResp{revision{2, 1}, revision{2, 0}, 2, nil},
  78. revision{2, 1},
  79. newTestKeyBytes(revision{3, 0}, false),
  80. storagepb.KeyValue{
  81. Key: []byte("foo"),
  82. Value: []byte("bar"),
  83. CreateRevision: 2,
  84. ModRevision: 3,
  85. Version: 3,
  86. },
  87. revision{3, 0},
  88. },
  89. }
  90. for i, tt := range tests {
  91. s := newFakeStore()
  92. b := s.b.(*fakeBackend)
  93. fi := s.kvindex.(*fakeIndex)
  94. s.currentRev = tt.rev
  95. s.tx = b.BatchTx()
  96. fi.indexGetRespc <- tt.r
  97. s.put([]byte("foo"), []byte("bar"))
  98. data, err := tt.wkv.Marshal()
  99. if err != nil {
  100. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  101. }
  102. wact := []testutil.Action{
  103. {"put", []interface{}{keyBucketName, tt.wkey, data}},
  104. }
  105. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  106. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  107. }
  108. wact = []testutil.Action{
  109. {"get", []interface{}{[]byte("foo"), tt.wputrev.main}},
  110. {"put", []interface{}{[]byte("foo"), tt.wputrev}},
  111. }
  112. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  113. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  114. }
  115. if s.currentRev != tt.wrev {
  116. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  117. }
  118. }
  119. }
  120. func TestStoreRange(t *testing.T) {
  121. key := newTestKeyBytes(revision{2, 0}, false)
  122. kv := storagepb.KeyValue{
  123. Key: []byte("foo"),
  124. Value: []byte("bar"),
  125. CreateRevision: 1,
  126. ModRevision: 2,
  127. Version: 1,
  128. }
  129. kvb, err := kv.Marshal()
  130. if err != nil {
  131. t.Fatal(err)
  132. }
  133. currev := revision{1, 1}
  134. wrev := int64(2)
  135. tests := []struct {
  136. idxr indexRangeResp
  137. r rangeResp
  138. }{
  139. {
  140. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  141. rangeResp{[][]byte{key}, [][]byte{kvb}},
  142. },
  143. {
  144. indexRangeResp{[][]byte{[]byte("foo"), []byte("foo1")}, []revision{{2, 0}, {3, 0}}},
  145. rangeResp{[][]byte{key}, [][]byte{kvb}},
  146. },
  147. }
  148. for i, tt := range tests {
  149. s := newFakeStore()
  150. b := s.b.(*fakeBackend)
  151. fi := s.kvindex.(*fakeIndex)
  152. s.currentRev = currev
  153. s.tx = b.BatchTx()
  154. b.tx.rangeRespc <- tt.r
  155. fi.indexRangeRespc <- tt.idxr
  156. kvs, rev, err := s.rangeKeys([]byte("foo"), []byte("goo"), 1, 0)
  157. if err != nil {
  158. t.Errorf("#%d: err = %v, want nil", i, err)
  159. }
  160. if w := []storagepb.KeyValue{kv}; !reflect.DeepEqual(kvs, w) {
  161. t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, w)
  162. }
  163. if rev != wrev {
  164. t.Errorf("#%d: rev = %d, want %d", i, rev, wrev)
  165. }
  166. wstart, wend := revBytesRange(tt.idxr.revs[0])
  167. wact := []testutil.Action{
  168. {"range", []interface{}{keyBucketName, wstart, wend, int64(0)}},
  169. }
  170. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  171. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  172. }
  173. wact = []testutil.Action{
  174. {"range", []interface{}{[]byte("foo"), []byte("goo"), wrev}},
  175. }
  176. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  177. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  178. }
  179. if s.currentRev != currev {
  180. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, currev)
  181. }
  182. }
  183. }
  184. func TestStoreDeleteRange(t *testing.T) {
  185. tests := []struct {
  186. rev revision
  187. r indexRangeResp
  188. wkey []byte
  189. wrev revision
  190. wrrev int64
  191. wdelrev revision
  192. }{
  193. {
  194. revision{2, 0},
  195. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  196. newTestKeyBytes(revision{3, 0}, true),
  197. revision{2, 1},
  198. 2,
  199. revision{3, 0},
  200. },
  201. {
  202. revision{2, 1},
  203. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  204. newTestKeyBytes(revision{3, 1}, true),
  205. revision{2, 2},
  206. 3,
  207. revision{3, 1},
  208. },
  209. }
  210. for i, tt := range tests {
  211. s := newFakeStore()
  212. b := s.b.(*fakeBackend)
  213. fi := s.kvindex.(*fakeIndex)
  214. s.currentRev = tt.rev
  215. s.tx = b.BatchTx()
  216. fi.indexRangeRespc <- tt.r
  217. n := s.deleteRange([]byte("foo"), []byte("goo"))
  218. if n != 1 {
  219. t.Errorf("#%d: n = %d, want 1", i, n)
  220. }
  221. data, err := (&storagepb.KeyValue{
  222. Key: []byte("foo"),
  223. }).Marshal()
  224. if err != nil {
  225. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  226. }
  227. wact := []testutil.Action{
  228. {"put", []interface{}{keyBucketName, tt.wkey, data}},
  229. }
  230. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  231. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  232. }
  233. wact = []testutil.Action{
  234. {"range", []interface{}{[]byte("foo"), []byte("goo"), tt.wrrev}},
  235. {"tombstone", []interface{}{[]byte("foo"), tt.wdelrev}},
  236. }
  237. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  238. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  239. }
  240. if s.currentRev != tt.wrev {
  241. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  242. }
  243. }
  244. }
  245. func TestStoreRangeHistory(t *testing.T) {
  246. key := newTestKeyBytes(revision{2, 0}, false)
  247. kv := storagepb.KeyValue{
  248. Key: []byte("foo"),
  249. Value: []byte("bar"),
  250. CreateRevision: 1,
  251. ModRevision: 2,
  252. Version: 1,
  253. }
  254. kvb, err := kv.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{key}, [][]byte{kvb}},
  266. },
  267. {
  268. indexRangeEventsResp{[]revision{{2, 0}, {3, 0}}},
  269. rangeResp{[][]byte{key}, [][]byte{kvb}},
  270. },
  271. }
  272. for i, tt := range tests {
  273. s := newFakeStore()
  274. b := s.b.(*fakeBackend)
  275. fi := s.kvindex.(*fakeIndex)
  276. s.currentRev = currev
  277. fi.indexRangeEventsRespc <- tt.idxr
  278. b.tx.rangeRespc <- tt.r
  279. keys, kvs, _, err := s.RangeHistory([]byte("foo"), []byte("goo"), 1, 1)
  280. if err != nil {
  281. t.Errorf("#%d: err = %v, want nil", i, err)
  282. }
  283. if w := [][]byte{key}; !reflect.DeepEqual(keys, w) {
  284. t.Errorf("#%d: keys = %+v, want %+v", i, keys, w)
  285. }
  286. if w := []storagepb.KeyValue{kv}; !reflect.DeepEqual(kvs, w) {
  287. t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, w)
  288. }
  289. wact := []testutil.Action{
  290. {"rangeEvents", []interface{}{[]byte("foo"), []byte("goo"), int64(1)}},
  291. }
  292. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  293. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  294. }
  295. wstart, wend := revBytesRange(tt.idxr.revs[0])
  296. wact = []testutil.Action{
  297. {"range", []interface{}{keyBucketName, wstart, wend, int64(0)}},
  298. }
  299. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  300. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  301. }
  302. if s.currentRev != currev {
  303. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, currev)
  304. }
  305. }
  306. }
  307. func TestStoreCompact(t *testing.T) {
  308. s := newFakeStore()
  309. b := s.b.(*fakeBackend)
  310. fi := s.kvindex.(*fakeIndex)
  311. s.currentRev = revision{3, 0}
  312. fi.indexCompactRespc <- map[revision]struct{}{revision{1, 0}: {}}
  313. key1 := newTestKeyBytes(revision{1, 0}, false)
  314. key2 := newTestKeyBytes(revision{2, 0}, false)
  315. b.tx.rangeRespc <- rangeResp{[][]byte{key1, key2}, nil}
  316. s.Compact(3)
  317. s.wg.Wait()
  318. if s.compactMainRev != 3 {
  319. t.Errorf("compact main rev = %d, want 3", s.compactMainRev)
  320. }
  321. end := make([]byte, 8)
  322. binary.BigEndian.PutUint64(end, uint64(4))
  323. wact := []testutil.Action{
  324. {"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestRevBytes(revision{3, 0})}},
  325. {"range", []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
  326. {"delete", []interface{}{keyBucketName, key2}},
  327. {"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestRevBytes(revision{3, 0})}},
  328. }
  329. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  330. t.Errorf("tx actions = %+v, want %+v", g, wact)
  331. }
  332. wact = []testutil.Action{
  333. {"compact", []interface{}{int64(3)}},
  334. }
  335. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  336. t.Errorf("index action = %+v, want %+v", g, wact)
  337. }
  338. }
  339. func TestStoreRestore(t *testing.T) {
  340. s := newFakeStore()
  341. b := s.b.(*fakeBackend)
  342. fi := s.kvindex.(*fakeIndex)
  343. putkey := newTestKeyBytes(revision{3, 0}, false)
  344. putkv := storagepb.KeyValue{
  345. Key: []byte("foo"),
  346. Value: []byte("bar"),
  347. CreateRevision: 3,
  348. ModRevision: 3,
  349. Version: 1,
  350. }
  351. putkvb, err := putkv.Marshal()
  352. if err != nil {
  353. t.Fatal(err)
  354. }
  355. delkey := newTestKeyBytes(revision{4, 0}, true)
  356. delkv := storagepb.KeyValue{
  357. Key: []byte("foo"),
  358. }
  359. delkvb, err := delkv.Marshal()
  360. if err != nil {
  361. t.Fatal(err)
  362. }
  363. b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{2, 0})}}
  364. b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
  365. b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{2, 0})}}
  366. s.Restore()
  367. if s.compactMainRev != 2 {
  368. t.Errorf("compact rev = %d, want 4", s.compactMainRev)
  369. }
  370. wrev := revision{4, 0}
  371. if !reflect.DeepEqual(s.currentRev, wrev) {
  372. t.Errorf("current rev = %v, want %v", s.currentRev, wrev)
  373. }
  374. wact := []testutil.Action{
  375. {"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
  376. {"range", []interface{}{keyBucketName, newTestRevBytes(revision{}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(0)}},
  377. {"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
  378. }
  379. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  380. t.Errorf("tx actions = %+v, want %+v", g, wact)
  381. }
  382. wact = []testutil.Action{
  383. {"restore", []interface{}{[]byte("foo"), revision{3, 0}, revision{3, 0}, int64(1)}},
  384. {"tombstone", []interface{}{[]byte("foo"), revision{4, 0}}},
  385. }
  386. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  387. t.Errorf("index action = %+v, want %+v", g, wact)
  388. }
  389. }
  390. // tests end parameter works well
  391. func TestStoreRangeHistoryEnd(t *testing.T) {
  392. s := newStore(tmpPath)
  393. defer cleanup(s, tmpPath)
  394. s.Put([]byte("foo"), []byte("bar"))
  395. s.Put([]byte("foo1"), []byte("bar1"))
  396. s.Put([]byte("foo2"), []byte("bar2"))
  397. keys := [][]byte{
  398. newTestKeyBytes(revision{1, 0}, false),
  399. newTestKeyBytes(revision{2, 0}, false),
  400. newTestKeyBytes(revision{3, 0}, false),
  401. }
  402. kvs := []storagepb.KeyValue{
  403. {Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
  404. {Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
  405. {Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
  406. }
  407. tests := []struct {
  408. key, end []byte
  409. wkeys [][]byte
  410. wkvs []storagepb.KeyValue
  411. }{
  412. // get no keys
  413. {
  414. []byte("doo"), []byte("foo"),
  415. nil, nil,
  416. },
  417. // get no keys when key == end
  418. {
  419. []byte("foo"), []byte("foo"),
  420. nil, nil,
  421. },
  422. // get no keys when ranging single key
  423. {
  424. []byte("doo"), nil,
  425. nil, nil,
  426. },
  427. // get all keys
  428. {
  429. []byte("foo"), []byte("foo3"),
  430. keys, kvs,
  431. },
  432. // get partial keys
  433. {
  434. []byte("foo"), []byte("foo1"),
  435. keys[:1], kvs[:1],
  436. },
  437. // get single key
  438. {
  439. []byte("foo"), nil,
  440. keys[:1], kvs[:1],
  441. },
  442. }
  443. for i, tt := range tests {
  444. keys, kvs, rev, err := s.RangeHistory(tt.key, tt.end, 0, 1)
  445. if err != nil {
  446. t.Fatal(err)
  447. }
  448. if rev != 4 {
  449. t.Errorf("#%d: rev = %d, want %d", i, rev, 4)
  450. }
  451. if !reflect.DeepEqual(keys, tt.wkeys) {
  452. t.Errorf("#%d: actions = %+v, want %+v", i, keys, tt.wkeys)
  453. }
  454. if !reflect.DeepEqual(kvs, tt.wkvs) {
  455. t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
  456. }
  457. }
  458. }
  459. func TestStoreRangeHistoryRev(t *testing.T) {
  460. s := newStore(tmpPath)
  461. defer cleanup(s, tmpPath)
  462. s.Put([]byte("foo"), []byte("bar"))
  463. s.DeleteRange([]byte("foo"), nil)
  464. s.Put([]byte("foo"), []byte("bar"))
  465. s.Put([]byte("unrelated"), []byte("unrelated"))
  466. keys := [][]byte{
  467. newTestKeyBytes(revision{1, 0}, false),
  468. newTestKeyBytes(revision{2, 0}, true),
  469. newTestKeyBytes(revision{3, 0}, false),
  470. }
  471. kvs := []storagepb.KeyValue{
  472. {Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
  473. {Key: []byte("foo")},
  474. {Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
  475. }
  476. tests := []struct {
  477. start int64
  478. wkeys [][]byte
  479. wkvs []storagepb.KeyValue
  480. wnext int64
  481. }{
  482. {0, keys, kvs, 5},
  483. {1, keys, kvs, 5},
  484. {3, keys[2:], kvs[2:], 5},
  485. }
  486. for i, tt := range tests {
  487. keys, kvs, next, err := s.RangeHistory([]byte("foo"), nil, 0, tt.start)
  488. if err != nil {
  489. t.Fatal(err)
  490. }
  491. if !reflect.DeepEqual(keys, tt.wkeys) {
  492. t.Errorf("#%d: acts = %+v, want %+v", i, keys, tt.wkeys)
  493. }
  494. if !reflect.DeepEqual(kvs, tt.wkvs) {
  495. t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
  496. }
  497. if next != tt.wnext {
  498. t.Errorf("#%d: next = %d, want %d", i, next, tt.wnext)
  499. }
  500. }
  501. }
  502. func TestStoreRangeHistoryBad(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.RangeHistory([]byte("foo"), nil, 0, tt.rev)
  523. if err != tt.werr {
  524. t.Errorf("#%d: error = %v, want %v", i, err, tt.werr)
  525. }
  526. }
  527. }
  528. func TestStoreRangeHistoryLimit(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. keys := [][]byte{
  535. newTestKeyBytes(revision{1, 0}, false),
  536. newTestKeyBytes(revision{2, 0}, true),
  537. newTestKeyBytes(revision{3, 0}, false),
  538. }
  539. kvs := []storagepb.KeyValue{
  540. {Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
  541. {Key: []byte("foo")},
  542. {Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
  543. }
  544. tests := []struct {
  545. limit int64
  546. wkeys [][]byte
  547. wkvs []storagepb.KeyValue
  548. }{
  549. // no limit
  550. {-1, keys, kvs},
  551. // no limit
  552. {0, keys, kvs},
  553. {1, keys[:1], kvs[:1]},
  554. {2, keys[:2], kvs[:2]},
  555. {3, keys, kvs},
  556. {100, keys, kvs},
  557. }
  558. for i, tt := range tests {
  559. keys, kvs, _, err := s.RangeHistory([]byte("foo"), nil, tt.limit, 1)
  560. if err != nil {
  561. t.Fatalf("#%d: range error (%v)", i, err)
  562. }
  563. if !reflect.DeepEqual(keys, tt.wkeys) {
  564. t.Errorf("#%d: acts = %+v, want %+v", i, keys, tt.wkeys)
  565. }
  566. if !reflect.DeepEqual(kvs, tt.wkvs) {
  567. t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
  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. // The disk compaction is done asynchronously and requires more time on slow disk.
  597. // try 5 times for CI with slow IO.
  598. for i := 0; i < 5; i++ {
  599. tx = s1.b.BatchTx()
  600. tx.Lock()
  601. ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
  602. tx.Unlock()
  603. if len(ks) != 0 {
  604. time.Sleep(100 * time.Millisecond)
  605. continue
  606. }
  607. return
  608. }
  609. t.Errorf("key for rev %+v still exists, want deleted", bytesToRev(revbytes))
  610. }
  611. func TestTxnPut(t *testing.T) {
  612. // assign arbitrary size
  613. bytesN := 30
  614. sliceN := 100
  615. keys := createBytesSlice(bytesN, sliceN)
  616. vals := createBytesSlice(bytesN, sliceN)
  617. s := newStore(tmpPath)
  618. defer cleanup(s, tmpPath)
  619. for i := 0; i < sliceN; i++ {
  620. id := s.TxnBegin()
  621. base := int64(i + 1)
  622. rev, err := s.TxnPut(id, keys[i], vals[i])
  623. if err != nil {
  624. t.Error("txn put error")
  625. }
  626. if rev != base {
  627. t.Errorf("#%d: rev = %d, want %d", i, rev, base)
  628. }
  629. s.TxnEnd(id)
  630. }
  631. }
  632. func TestTxnBlockBackendForceCommit(t *testing.T) {
  633. s := newStore(tmpPath)
  634. defer os.Remove(tmpPath)
  635. id := s.TxnBegin()
  636. done := make(chan struct{})
  637. go func() {
  638. s.b.ForceCommit()
  639. done <- struct{}{}
  640. }()
  641. select {
  642. case <-done:
  643. t.Fatalf("failed to block ForceCommit")
  644. case <-time.After(100 * time.Millisecond):
  645. }
  646. s.TxnEnd(id)
  647. select {
  648. case <-done:
  649. case <-time.After(time.Second):
  650. t.Fatalf("failed to execute ForceCommit")
  651. }
  652. }
  653. func newTestRevBytes(rev revision) []byte {
  654. bytes := newRevBytes()
  655. revToBytes(rev, bytes)
  656. return bytes
  657. }
  658. func newTestKeyBytes(rev revision, tombstone bool) []byte {
  659. bytes := newRevBytes()
  660. revToBytes(rev, bytes)
  661. if tombstone {
  662. bytes = appendMarkTombstone(bytes)
  663. }
  664. return bytes
  665. }
  666. func newFakeStore() *store {
  667. b := &fakeBackend{&fakeBatchTx{rangeRespc: make(chan rangeResp, 5)}}
  668. fi := &fakeIndex{
  669. indexGetRespc: make(chan indexGetResp, 1),
  670. indexRangeRespc: make(chan indexRangeResp, 1),
  671. indexRangeEventsRespc: make(chan indexRangeEventsResp, 1),
  672. indexCompactRespc: make(chan map[revision]struct{}, 1),
  673. }
  674. return &store{
  675. b: b,
  676. kvindex: fi,
  677. currentRev: revision{},
  678. compactMainRev: -1,
  679. }
  680. }
  681. type rangeResp struct {
  682. keys [][]byte
  683. vals [][]byte
  684. }
  685. type fakeBatchTx struct {
  686. testutil.Recorder
  687. rangeRespc chan rangeResp
  688. }
  689. func (b *fakeBatchTx) Lock() {}
  690. func (b *fakeBatchTx) Unlock() {}
  691. func (b *fakeBatchTx) UnsafeCreateBucket(name []byte) {}
  692. func (b *fakeBatchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
  693. b.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{bucketName, key, value}})
  694. }
  695. func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
  696. b.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{bucketName, key, endKey, limit}})
  697. r := <-b.rangeRespc
  698. return r.keys, r.vals
  699. }
  700. func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
  701. b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
  702. }
  703. func (b *fakeBatchTx) Commit() {}
  704. func (b *fakeBatchTx) CommitAndStop() {}
  705. type fakeBackend struct {
  706. tx *fakeBatchTx
  707. }
  708. func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
  709. func (b *fakeBackend) Hash() (uint32, error) { return 0, nil }
  710. func (b *fakeBackend) Size() int64 { return 0 }
  711. func (b *fakeBackend) Snapshot() backend.Snapshot { return nil }
  712. func (b *fakeBackend) ForceCommit() {}
  713. func (b *fakeBackend) Close() error { return nil }
  714. type indexGetResp struct {
  715. rev revision
  716. created revision
  717. ver int64
  718. err error
  719. }
  720. type indexRangeResp struct {
  721. keys [][]byte
  722. revs []revision
  723. }
  724. type indexRangeEventsResp struct {
  725. revs []revision
  726. }
  727. type fakeIndex struct {
  728. testutil.Recorder
  729. indexGetRespc chan indexGetResp
  730. indexRangeRespc chan indexRangeResp
  731. indexRangeEventsRespc chan indexRangeEventsResp
  732. indexCompactRespc chan map[revision]struct{}
  733. }
  734. func (i *fakeIndex) Get(key []byte, atRev int64) (rev, created revision, ver int64, err error) {
  735. i.Recorder.Record(testutil.Action{Name: "get", Params: []interface{}{key, atRev}})
  736. r := <-i.indexGetRespc
  737. return r.rev, r.created, r.ver, r.err
  738. }
  739. func (i *fakeIndex) Range(key, end []byte, atRev int64) ([][]byte, []revision) {
  740. i.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{key, end, atRev}})
  741. r := <-i.indexRangeRespc
  742. return r.keys, r.revs
  743. }
  744. func (i *fakeIndex) Put(key []byte, rev revision) {
  745. i.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{key, rev}})
  746. }
  747. func (i *fakeIndex) Restore(key []byte, created, modified revision, ver int64) {
  748. i.Recorder.Record(testutil.Action{Name: "restore", Params: []interface{}{key, created, modified, ver}})
  749. }
  750. func (i *fakeIndex) Tombstone(key []byte, rev revision) error {
  751. i.Recorder.Record(testutil.Action{Name: "tombstone", Params: []interface{}{key, rev}})
  752. return nil
  753. }
  754. func (i *fakeIndex) RangeSince(key, end []byte, rev int64) []revision {
  755. i.Recorder.Record(testutil.Action{Name: "rangeEvents", Params: []interface{}{key, end, rev}})
  756. r := <-i.indexRangeEventsRespc
  757. return r.revs
  758. }
  759. func (i *fakeIndex) Compact(rev int64) map[revision]struct{} {
  760. i.Recorder.Record(testutil.Action{Name: "compact", Params: []interface{}{rev}})
  761. return <-i.indexCompactRespc
  762. }
  763. func (i *fakeIndex) Equal(b index) bool { return false }
  764. func createBytesSlice(bytesN, sliceN int) [][]byte {
  765. rs := [][]byte{}
  766. for len(rs) != sliceN {
  767. mrand.Seed(time.Now().UnixNano())
  768. v := make([]byte, bytesN)
  769. if _, err := rand.Read(v); err != nil {
  770. panic(err)
  771. }
  772. rs = append(rs, v)
  773. }
  774. return rs
  775. }