kvstore_test.go 22 KB

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