kvstore_test.go 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718
  1. // Copyright 2015 The etcd Authors
  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 mvcc
  15. import (
  16. "crypto/rand"
  17. "encoding/binary"
  18. "fmt"
  19. "math"
  20. mrand "math/rand"
  21. "os"
  22. "reflect"
  23. "testing"
  24. "time"
  25. "github.com/coreos/etcd/lease"
  26. "github.com/coreos/etcd/mvcc/backend"
  27. "github.com/coreos/etcd/mvcc/mvccpb"
  28. "github.com/coreos/etcd/pkg/schedule"
  29. "github.com/coreos/etcd/pkg/testutil"
  30. )
  31. func TestStoreRev(t *testing.T) {
  32. b, tmpPath := backend.NewDefaultTmpBackend()
  33. s := NewStore(b, &lease.FakeLessor{}, nil)
  34. defer s.Close()
  35. defer os.Remove(tmpPath)
  36. for i := 1; i <= 3; i++ {
  37. s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
  38. if r := s.Rev(); r != int64(i+1) {
  39. t.Errorf("#%d: rev = %d, want %d", i, r, i+1)
  40. }
  41. }
  42. }
  43. func TestStorePut(t *testing.T) {
  44. kv := mvccpb.KeyValue{
  45. Key: []byte("foo"),
  46. Value: []byte("bar"),
  47. CreateRevision: 1,
  48. ModRevision: 2,
  49. Version: 1,
  50. }
  51. kvb, err := kv.Marshal()
  52. if err != nil {
  53. t.Fatal(err)
  54. }
  55. tests := []struct {
  56. rev revision
  57. r indexGetResp
  58. rr *rangeResp
  59. wrev revision
  60. wkey []byte
  61. wkv mvccpb.KeyValue
  62. wputrev revision
  63. }{
  64. {
  65. revision{1, 0},
  66. indexGetResp{revision{}, revision{}, 0, ErrRevisionNotFound},
  67. nil,
  68. revision{2, 0},
  69. newTestKeyBytes(revision{2, 0}, false),
  70. mvccpb.KeyValue{
  71. Key: []byte("foo"),
  72. Value: []byte("bar"),
  73. CreateRevision: 2,
  74. ModRevision: 2,
  75. Version: 1,
  76. Lease: 1,
  77. },
  78. revision{2, 0},
  79. },
  80. {
  81. revision{1, 1},
  82. indexGetResp{revision{2, 0}, revision{2, 0}, 1, nil},
  83. &rangeResp{[][]byte{newTestKeyBytes(revision{2, 1}, false)}, [][]byte{kvb}},
  84. revision{2, 0},
  85. newTestKeyBytes(revision{2, 0}, false),
  86. mvccpb.KeyValue{
  87. Key: []byte("foo"),
  88. Value: []byte("bar"),
  89. CreateRevision: 2,
  90. ModRevision: 2,
  91. Version: 2,
  92. Lease: 2,
  93. },
  94. revision{2, 0},
  95. },
  96. {
  97. revision{2, 0},
  98. indexGetResp{revision{2, 1}, revision{2, 0}, 2, nil},
  99. &rangeResp{[][]byte{newTestKeyBytes(revision{2, 1}, false)}, [][]byte{kvb}},
  100. revision{3, 0},
  101. newTestKeyBytes(revision{3, 0}, false),
  102. mvccpb.KeyValue{
  103. Key: []byte("foo"),
  104. Value: []byte("bar"),
  105. CreateRevision: 2,
  106. ModRevision: 3,
  107. Version: 3,
  108. Lease: 3,
  109. },
  110. revision{3, 0},
  111. },
  112. }
  113. for i, tt := range tests {
  114. s := newFakeStore()
  115. b := s.b.(*fakeBackend)
  116. fi := s.kvindex.(*fakeIndex)
  117. s.currentRev = tt.rev.main
  118. fi.indexGetRespc <- tt.r
  119. if tt.rr != nil {
  120. b.tx.rangeRespc <- *tt.rr
  121. }
  122. s.Put([]byte("foo"), []byte("bar"), lease.LeaseID(i+1))
  123. data, err := tt.wkv.Marshal()
  124. if err != nil {
  125. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  126. }
  127. wact := []testutil.Action{
  128. {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
  129. }
  130. if tt.rr != nil {
  131. wact = []testutil.Action{
  132. {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
  133. }
  134. }
  135. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  136. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  137. }
  138. wact = []testutil.Action{
  139. {"get", []interface{}{[]byte("foo"), tt.wputrev.main}},
  140. {"put", []interface{}{[]byte("foo"), tt.wputrev}},
  141. }
  142. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  143. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  144. }
  145. if s.currentRev != tt.wrev.main {
  146. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  147. }
  148. s.Close()
  149. }
  150. }
  151. func TestStoreRange(t *testing.T) {
  152. key := newTestKeyBytes(revision{2, 0}, false)
  153. kv := mvccpb.KeyValue{
  154. Key: []byte("foo"),
  155. Value: []byte("bar"),
  156. CreateRevision: 1,
  157. ModRevision: 2,
  158. Version: 1,
  159. }
  160. kvb, err := kv.Marshal()
  161. if err != nil {
  162. t.Fatal(err)
  163. }
  164. wrev := int64(2)
  165. tests := []struct {
  166. idxr indexRangeResp
  167. r rangeResp
  168. }{
  169. {
  170. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  171. rangeResp{[][]byte{key}, [][]byte{kvb}},
  172. },
  173. {
  174. indexRangeResp{[][]byte{[]byte("foo"), []byte("foo1")}, []revision{{2, 0}, {3, 0}}},
  175. rangeResp{[][]byte{key}, [][]byte{kvb}},
  176. },
  177. }
  178. ro := RangeOptions{Limit: 1, Rev: 0, Count: false}
  179. for i, tt := range tests {
  180. s := newFakeStore()
  181. b := s.b.(*fakeBackend)
  182. fi := s.kvindex.(*fakeIndex)
  183. s.currentRev = 2
  184. b.tx.rangeRespc <- tt.r
  185. fi.indexRangeRespc <- tt.idxr
  186. ret, err := s.Range([]byte("foo"), []byte("goo"), ro)
  187. if err != nil {
  188. t.Errorf("#%d: err = %v, want nil", i, err)
  189. }
  190. if w := []mvccpb.KeyValue{kv}; !reflect.DeepEqual(ret.KVs, w) {
  191. t.Errorf("#%d: kvs = %+v, want %+v", i, ret.KVs, w)
  192. }
  193. if ret.Rev != wrev {
  194. t.Errorf("#%d: rev = %d, want %d", i, ret.Rev, wrev)
  195. }
  196. wstart, wend := revBytesRange(tt.idxr.revs[0])
  197. wact := []testutil.Action{
  198. {"range", []interface{}{keyBucketName, wstart, wend, int64(0)}},
  199. }
  200. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  201. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  202. }
  203. wact = []testutil.Action{
  204. {"range", []interface{}{[]byte("foo"), []byte("goo"), wrev}},
  205. }
  206. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  207. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  208. }
  209. if s.currentRev != 2 {
  210. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, 2)
  211. }
  212. s.Close()
  213. }
  214. }
  215. func TestStoreDeleteRange(t *testing.T) {
  216. key := newTestKeyBytes(revision{2, 0}, false)
  217. kv := mvccpb.KeyValue{
  218. Key: []byte("foo"),
  219. Value: []byte("bar"),
  220. CreateRevision: 1,
  221. ModRevision: 2,
  222. Version: 1,
  223. }
  224. kvb, err := kv.Marshal()
  225. if err != nil {
  226. t.Fatal(err)
  227. }
  228. tests := []struct {
  229. rev revision
  230. r indexRangeResp
  231. rr rangeResp
  232. wkey []byte
  233. wrev revision
  234. wrrev int64
  235. wdelrev revision
  236. }{
  237. {
  238. revision{2, 0},
  239. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  240. rangeResp{[][]byte{key}, [][]byte{kvb}},
  241. newTestKeyBytes(revision{3, 0}, true),
  242. revision{3, 0},
  243. 2,
  244. revision{3, 0},
  245. },
  246. }
  247. for i, tt := range tests {
  248. s := newFakeStore()
  249. b := s.b.(*fakeBackend)
  250. fi := s.kvindex.(*fakeIndex)
  251. s.currentRev = tt.rev.main
  252. fi.indexRangeRespc <- tt.r
  253. b.tx.rangeRespc <- tt.rr
  254. n, _ := s.DeleteRange([]byte("foo"), []byte("goo"))
  255. if n != 1 {
  256. t.Errorf("#%d: n = %d, want 1", i, n)
  257. }
  258. data, err := (&mvccpb.KeyValue{
  259. Key: []byte("foo"),
  260. }).Marshal()
  261. if err != nil {
  262. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  263. }
  264. wact := []testutil.Action{
  265. {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
  266. }
  267. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  268. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  269. }
  270. wact = []testutil.Action{
  271. {"range", []interface{}{[]byte("foo"), []byte("goo"), tt.wrrev}},
  272. {"tombstone", []interface{}{[]byte("foo"), tt.wdelrev}},
  273. }
  274. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  275. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  276. }
  277. if s.currentRev != tt.wrev.main {
  278. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  279. }
  280. }
  281. }
  282. func TestStoreCompact(t *testing.T) {
  283. s := newFakeStore()
  284. defer s.Close()
  285. b := s.b.(*fakeBackend)
  286. fi := s.kvindex.(*fakeIndex)
  287. s.currentRev = 3
  288. fi.indexCompactRespc <- map[revision]struct{}{{1, 0}: {}}
  289. key1 := newTestKeyBytes(revision{1, 0}, false)
  290. key2 := newTestKeyBytes(revision{2, 0}, false)
  291. b.tx.rangeRespc <- rangeResp{[][]byte{key1, key2}, nil}
  292. s.Compact(3)
  293. s.fifoSched.WaitFinish(1)
  294. if s.compactMainRev != 3 {
  295. t.Errorf("compact main rev = %d, want 3", s.compactMainRev)
  296. }
  297. end := make([]byte, 8)
  298. binary.BigEndian.PutUint64(end, uint64(4))
  299. wact := []testutil.Action{
  300. {"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestRevBytes(revision{3, 0})}},
  301. {"range", []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
  302. {"delete", []interface{}{keyBucketName, key2}},
  303. {"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestRevBytes(revision{3, 0})}},
  304. }
  305. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  306. t.Errorf("tx actions = %+v, want %+v", g, wact)
  307. }
  308. wact = []testutil.Action{
  309. {"compact", []interface{}{int64(3)}},
  310. }
  311. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  312. t.Errorf("index action = %+v, want %+v", g, wact)
  313. }
  314. }
  315. func TestStoreRestore(t *testing.T) {
  316. s := newFakeStore()
  317. b := s.b.(*fakeBackend)
  318. fi := s.kvindex.(*fakeIndex)
  319. putkey := newTestKeyBytes(revision{3, 0}, false)
  320. putkv := mvccpb.KeyValue{
  321. Key: []byte("foo"),
  322. Value: []byte("bar"),
  323. CreateRevision: 4,
  324. ModRevision: 4,
  325. Version: 1,
  326. }
  327. putkvb, err := putkv.Marshal()
  328. if err != nil {
  329. t.Fatal(err)
  330. }
  331. delkey := newTestKeyBytes(revision{5, 0}, true)
  332. delkv := mvccpb.KeyValue{
  333. Key: []byte("foo"),
  334. }
  335. delkvb, err := delkv.Marshal()
  336. if err != nil {
  337. t.Fatal(err)
  338. }
  339. b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
  340. b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
  341. b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
  342. b.tx.rangeRespc <- rangeResp{nil, nil}
  343. s.restore()
  344. if s.compactMainRev != 3 {
  345. t.Errorf("compact rev = %d, want 5", s.compactMainRev)
  346. }
  347. if s.currentRev != 5 {
  348. t.Errorf("current rev = %v, want 5", s.currentRev)
  349. }
  350. wact := []testutil.Action{
  351. {"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
  352. {"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
  353. {"range", []interface{}{keyBucketName, newTestRevBytes(revision{1, 0}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(restoreChunkKeys)}},
  354. }
  355. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  356. t.Errorf("tx actions = %+v, want %+v", g, wact)
  357. }
  358. gens := []generation{
  359. {created: revision{4, 0}, ver: 2, revs: []revision{{3, 0}, {5, 0}}},
  360. {created: revision{0, 0}, ver: 0, revs: nil},
  361. }
  362. ki := &keyIndex{key: []byte("foo"), modified: revision{5, 0}, generations: gens}
  363. wact = []testutil.Action{
  364. {"keyIndex", []interface{}{ki}},
  365. {"insert", []interface{}{ki}},
  366. }
  367. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  368. t.Errorf("index action = %+v, want %+v", g, wact)
  369. }
  370. }
  371. func TestRestoreDelete(t *testing.T) {
  372. oldChunk := restoreChunkKeys
  373. restoreChunkKeys = mrand.Intn(3) + 2
  374. defer func() { restoreChunkKeys = oldChunk }()
  375. b, tmpPath := backend.NewDefaultTmpBackend()
  376. s := NewStore(b, &lease.FakeLessor{}, nil)
  377. defer os.Remove(tmpPath)
  378. keys := make(map[string]struct{})
  379. for i := 0; i < 20; i++ {
  380. ks := fmt.Sprintf("foo-%d", i)
  381. k := []byte(ks)
  382. s.Put(k, []byte("bar"), lease.NoLease)
  383. keys[ks] = struct{}{}
  384. switch mrand.Intn(3) {
  385. case 0:
  386. // put random key from past via random range on map
  387. ks = fmt.Sprintf("foo-%d", mrand.Intn(i+1))
  388. s.Put([]byte(ks), []byte("baz"), lease.NoLease)
  389. keys[ks] = struct{}{}
  390. case 1:
  391. // delete random key via random range on map
  392. for k := range keys {
  393. s.DeleteRange([]byte(k), nil)
  394. delete(keys, k)
  395. break
  396. }
  397. }
  398. }
  399. s.Close()
  400. s = NewStore(b, &lease.FakeLessor{}, nil)
  401. defer s.Close()
  402. for i := 0; i < 20; i++ {
  403. ks := fmt.Sprintf("foo-%d", i)
  404. r, err := s.Range([]byte(ks), nil, RangeOptions{})
  405. if err != nil {
  406. t.Fatal(err)
  407. }
  408. if _, ok := keys[ks]; ok {
  409. if len(r.KVs) == 0 {
  410. t.Errorf("#%d: expected %q, got deleted", i, ks)
  411. }
  412. } else if len(r.KVs) != 0 {
  413. t.Errorf("#%d: expected deleted, got %q", i, ks)
  414. }
  415. }
  416. }
  417. func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
  418. b, tmpPath := backend.NewDefaultTmpBackend()
  419. s0 := NewStore(b, &lease.FakeLessor{}, nil)
  420. defer os.Remove(tmpPath)
  421. s0.Put([]byte("foo"), []byte("bar"), lease.NoLease)
  422. s0.Put([]byte("foo"), []byte("bar1"), lease.NoLease)
  423. s0.Put([]byte("foo"), []byte("bar2"), lease.NoLease)
  424. // write scheduled compaction, but not do compaction
  425. rbytes := newRevBytes()
  426. revToBytes(revision{main: 2}, rbytes)
  427. tx := s0.b.BatchTx()
  428. tx.Lock()
  429. tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
  430. tx.Unlock()
  431. s0.Close()
  432. s1 := NewStore(b, &lease.FakeLessor{}, nil)
  433. // wait for scheduled compaction to be finished
  434. time.Sleep(100 * time.Millisecond)
  435. if _, err := s1.Range([]byte("foo"), nil, RangeOptions{Rev: 1}); err != ErrCompacted {
  436. t.Errorf("range on compacted rev error = %v, want %v", err, ErrCompacted)
  437. }
  438. // check the key in backend is deleted
  439. revbytes := newRevBytes()
  440. revToBytes(revision{main: 1}, revbytes)
  441. // The disk compaction is done asynchronously and requires more time on slow disk.
  442. // try 5 times for CI with slow IO.
  443. for i := 0; i < 5; i++ {
  444. tx = s1.b.BatchTx()
  445. tx.Lock()
  446. ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
  447. tx.Unlock()
  448. if len(ks) != 0 {
  449. time.Sleep(100 * time.Millisecond)
  450. continue
  451. }
  452. return
  453. }
  454. t.Errorf("key for rev %+v still exists, want deleted", bytesToRev(revbytes))
  455. }
  456. func TestTxnPut(t *testing.T) {
  457. // assign arbitrary size
  458. bytesN := 30
  459. sliceN := 100
  460. keys := createBytesSlice(bytesN, sliceN)
  461. vals := createBytesSlice(bytesN, sliceN)
  462. b, tmpPath := backend.NewDefaultTmpBackend()
  463. s := NewStore(b, &lease.FakeLessor{}, nil)
  464. defer cleanup(s, b, tmpPath)
  465. for i := 0; i < sliceN; i++ {
  466. txn := s.Write()
  467. base := int64(i + 2)
  468. if rev := txn.Put(keys[i], vals[i], lease.NoLease); rev != base {
  469. t.Errorf("#%d: rev = %d, want %d", i, rev, base)
  470. }
  471. txn.End()
  472. }
  473. }
  474. func TestTxnBlockBackendForceCommit(t *testing.T) {
  475. b, tmpPath := backend.NewDefaultTmpBackend()
  476. s := NewStore(b, &lease.FakeLessor{}, nil)
  477. defer os.Remove(tmpPath)
  478. txn := s.Read()
  479. done := make(chan struct{})
  480. go func() {
  481. s.b.ForceCommit()
  482. done <- struct{}{}
  483. }()
  484. select {
  485. case <-done:
  486. t.Fatalf("failed to block ForceCommit")
  487. case <-time.After(100 * time.Millisecond):
  488. }
  489. txn.End()
  490. select {
  491. case <-done:
  492. case <-time.After(5 * time.Second): // wait 5 seconds for CI with slow IO
  493. testutil.FatalStack(t, "failed to execute ForceCommit")
  494. }
  495. }
  496. // TODO: test attach key to lessor
  497. func newTestRevBytes(rev revision) []byte {
  498. bytes := newRevBytes()
  499. revToBytes(rev, bytes)
  500. return bytes
  501. }
  502. func newTestKeyBytes(rev revision, tombstone bool) []byte {
  503. bytes := newRevBytes()
  504. revToBytes(rev, bytes)
  505. if tombstone {
  506. bytes = appendMarkTombstone(bytes)
  507. }
  508. return bytes
  509. }
  510. func newFakeStore() *store {
  511. b := &fakeBackend{&fakeBatchTx{
  512. Recorder: &testutil.RecorderBuffered{},
  513. rangeRespc: make(chan rangeResp, 5)}}
  514. fi := &fakeIndex{
  515. Recorder: &testutil.RecorderBuffered{},
  516. indexGetRespc: make(chan indexGetResp, 1),
  517. indexRangeRespc: make(chan indexRangeResp, 1),
  518. indexRangeEventsRespc: make(chan indexRangeEventsResp, 1),
  519. indexCompactRespc: make(chan map[revision]struct{}, 1),
  520. }
  521. s := &store{
  522. b: b,
  523. le: &lease.FakeLessor{},
  524. kvindex: fi,
  525. currentRev: 0,
  526. compactMainRev: -1,
  527. fifoSched: schedule.NewFIFOScheduler(),
  528. stopc: make(chan struct{}),
  529. }
  530. s.ReadView, s.WriteView = &readView{s}, &writeView{s}
  531. return s
  532. }
  533. type rangeResp struct {
  534. keys [][]byte
  535. vals [][]byte
  536. }
  537. type fakeBatchTx struct {
  538. testutil.Recorder
  539. rangeRespc chan rangeResp
  540. }
  541. func (b *fakeBatchTx) Lock() {}
  542. func (b *fakeBatchTx) Unlock() {}
  543. func (b *fakeBatchTx) UnsafeCreateBucket(name []byte) {}
  544. func (b *fakeBatchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
  545. b.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{bucketName, key, value}})
  546. }
  547. func (b *fakeBatchTx) UnsafeSeqPut(bucketName []byte, key []byte, value []byte) {
  548. b.Recorder.Record(testutil.Action{Name: "seqput", Params: []interface{}{bucketName, key, value}})
  549. }
  550. func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
  551. b.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{bucketName, key, endKey, limit}})
  552. r := <-b.rangeRespc
  553. return r.keys, r.vals
  554. }
  555. func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
  556. b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
  557. }
  558. func (b *fakeBatchTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
  559. return nil
  560. }
  561. func (b *fakeBatchTx) Commit() {}
  562. func (b *fakeBatchTx) CommitAndStop() {}
  563. type fakeBackend struct {
  564. tx *fakeBatchTx
  565. }
  566. func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
  567. func (b *fakeBackend) ReadTx() backend.ReadTx { return b.tx }
  568. func (b *fakeBackend) Hash(ignores map[backend.IgnoreKey]struct{}) (uint32, error) { return 0, nil }
  569. func (b *fakeBackend) Size() int64 { return 0 }
  570. func (b *fakeBackend) SizeInUse() int64 { return 0 }
  571. func (b *fakeBackend) Snapshot() backend.Snapshot { return nil }
  572. func (b *fakeBackend) ForceCommit() {}
  573. func (b *fakeBackend) Defrag() error { return nil }
  574. func (b *fakeBackend) Close() error { return nil }
  575. type indexGetResp struct {
  576. rev revision
  577. created revision
  578. ver int64
  579. err error
  580. }
  581. type indexRangeResp struct {
  582. keys [][]byte
  583. revs []revision
  584. }
  585. type indexRangeEventsResp struct {
  586. revs []revision
  587. }
  588. type fakeIndex struct {
  589. testutil.Recorder
  590. indexGetRespc chan indexGetResp
  591. indexRangeRespc chan indexRangeResp
  592. indexRangeEventsRespc chan indexRangeEventsResp
  593. indexCompactRespc chan map[revision]struct{}
  594. }
  595. func (i *fakeIndex) Get(key []byte, atRev int64) (rev, created revision, ver int64, err error) {
  596. i.Recorder.Record(testutil.Action{Name: "get", Params: []interface{}{key, atRev}})
  597. r := <-i.indexGetRespc
  598. return r.rev, r.created, r.ver, r.err
  599. }
  600. func (i *fakeIndex) Range(key, end []byte, atRev int64) ([][]byte, []revision) {
  601. i.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{key, end, atRev}})
  602. r := <-i.indexRangeRespc
  603. return r.keys, r.revs
  604. }
  605. func (i *fakeIndex) Put(key []byte, rev revision) {
  606. i.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{key, rev}})
  607. }
  608. func (i *fakeIndex) Tombstone(key []byte, rev revision) error {
  609. i.Recorder.Record(testutil.Action{Name: "tombstone", Params: []interface{}{key, rev}})
  610. return nil
  611. }
  612. func (i *fakeIndex) RangeSince(key, end []byte, rev int64) []revision {
  613. i.Recorder.Record(testutil.Action{Name: "rangeEvents", Params: []interface{}{key, end, rev}})
  614. r := <-i.indexRangeEventsRespc
  615. return r.revs
  616. }
  617. func (i *fakeIndex) Compact(rev int64) map[revision]struct{} {
  618. i.Recorder.Record(testutil.Action{Name: "compact", Params: []interface{}{rev}})
  619. return <-i.indexCompactRespc
  620. }
  621. func (i *fakeIndex) Equal(b index) bool { return false }
  622. func (i *fakeIndex) Insert(ki *keyIndex) {
  623. i.Recorder.Record(testutil.Action{Name: "insert", Params: []interface{}{ki}})
  624. }
  625. func (i *fakeIndex) KeyIndex(ki *keyIndex) *keyIndex {
  626. i.Recorder.Record(testutil.Action{Name: "keyIndex", Params: []interface{}{ki}})
  627. return nil
  628. }
  629. func createBytesSlice(bytesN, sliceN int) [][]byte {
  630. rs := [][]byte{}
  631. for len(rs) != sliceN {
  632. v := make([]byte, bytesN)
  633. if _, err := rand.Read(v); err != nil {
  634. panic(err)
  635. }
  636. rs = append(rs, v)
  637. }
  638. return rs
  639. }