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 := newRevBytes()
  197. revToBytes(tt.idxr.revs[0], wstart)
  198. wact := []testutil.Action{
  199. {"range", []interface{}{keyBucketName, wstart, []byte(nil), int64(0)}},
  200. }
  201. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  202. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  203. }
  204. wact = []testutil.Action{
  205. {"range", []interface{}{[]byte("foo"), []byte("goo"), wrev}},
  206. }
  207. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  208. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  209. }
  210. if s.currentRev != 2 {
  211. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, 2)
  212. }
  213. s.Close()
  214. }
  215. }
  216. func TestStoreDeleteRange(t *testing.T) {
  217. key := newTestKeyBytes(revision{2, 0}, false)
  218. kv := mvccpb.KeyValue{
  219. Key: []byte("foo"),
  220. Value: []byte("bar"),
  221. CreateRevision: 1,
  222. ModRevision: 2,
  223. Version: 1,
  224. }
  225. kvb, err := kv.Marshal()
  226. if err != nil {
  227. t.Fatal(err)
  228. }
  229. tests := []struct {
  230. rev revision
  231. r indexRangeResp
  232. rr rangeResp
  233. wkey []byte
  234. wrev revision
  235. wrrev int64
  236. wdelrev revision
  237. }{
  238. {
  239. revision{2, 0},
  240. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  241. rangeResp{[][]byte{key}, [][]byte{kvb}},
  242. newTestKeyBytes(revision{3, 0}, true),
  243. revision{3, 0},
  244. 2,
  245. revision{3, 0},
  246. },
  247. }
  248. for i, tt := range tests {
  249. s := newFakeStore()
  250. b := s.b.(*fakeBackend)
  251. fi := s.kvindex.(*fakeIndex)
  252. s.currentRev = tt.rev.main
  253. fi.indexRangeRespc <- tt.r
  254. b.tx.rangeRespc <- tt.rr
  255. n, _ := s.DeleteRange([]byte("foo"), []byte("goo"))
  256. if n != 1 {
  257. t.Errorf("#%d: n = %d, want 1", i, n)
  258. }
  259. data, err := (&mvccpb.KeyValue{
  260. Key: []byte("foo"),
  261. }).Marshal()
  262. if err != nil {
  263. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  264. }
  265. wact := []testutil.Action{
  266. {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
  267. }
  268. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  269. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  270. }
  271. wact = []testutil.Action{
  272. {"range", []interface{}{[]byte("foo"), []byte("goo"), tt.wrrev}},
  273. {"tombstone", []interface{}{[]byte("foo"), tt.wdelrev}},
  274. }
  275. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  276. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  277. }
  278. if s.currentRev != tt.wrev.main {
  279. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  280. }
  281. }
  282. }
  283. func TestStoreCompact(t *testing.T) {
  284. s := newFakeStore()
  285. defer s.Close()
  286. b := s.b.(*fakeBackend)
  287. fi := s.kvindex.(*fakeIndex)
  288. s.currentRev = 3
  289. fi.indexCompactRespc <- map[revision]struct{}{{1, 0}: {}}
  290. key1 := newTestKeyBytes(revision{1, 0}, false)
  291. key2 := newTestKeyBytes(revision{2, 0}, false)
  292. b.tx.rangeRespc <- rangeResp{[][]byte{key1, key2}, nil}
  293. s.Compact(3)
  294. s.fifoSched.WaitFinish(1)
  295. if s.compactMainRev != 3 {
  296. t.Errorf("compact main rev = %d, want 3", s.compactMainRev)
  297. }
  298. end := make([]byte, 8)
  299. binary.BigEndian.PutUint64(end, uint64(4))
  300. wact := []testutil.Action{
  301. {"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestRevBytes(revision{3, 0})}},
  302. {"range", []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
  303. {"delete", []interface{}{keyBucketName, key2}},
  304. {"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestRevBytes(revision{3, 0})}},
  305. }
  306. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  307. t.Errorf("tx actions = %+v, want %+v", g, wact)
  308. }
  309. wact = []testutil.Action{
  310. {"compact", []interface{}{int64(3)}},
  311. }
  312. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  313. t.Errorf("index action = %+v, want %+v", g, wact)
  314. }
  315. }
  316. func TestStoreRestore(t *testing.T) {
  317. s := newFakeStore()
  318. b := s.b.(*fakeBackend)
  319. fi := s.kvindex.(*fakeIndex)
  320. putkey := newTestKeyBytes(revision{3, 0}, false)
  321. putkv := mvccpb.KeyValue{
  322. Key: []byte("foo"),
  323. Value: []byte("bar"),
  324. CreateRevision: 4,
  325. ModRevision: 4,
  326. Version: 1,
  327. }
  328. putkvb, err := putkv.Marshal()
  329. if err != nil {
  330. t.Fatal(err)
  331. }
  332. delkey := newTestKeyBytes(revision{5, 0}, true)
  333. delkv := mvccpb.KeyValue{
  334. Key: []byte("foo"),
  335. }
  336. delkvb, err := delkv.Marshal()
  337. if err != nil {
  338. t.Fatal(err)
  339. }
  340. b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
  341. b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
  342. b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
  343. b.tx.rangeRespc <- rangeResp{nil, nil}
  344. s.restore()
  345. if s.compactMainRev != 3 {
  346. t.Errorf("compact rev = %d, want 5", s.compactMainRev)
  347. }
  348. if s.currentRev != 5 {
  349. t.Errorf("current rev = %v, want 5", s.currentRev)
  350. }
  351. wact := []testutil.Action{
  352. {"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
  353. {"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
  354. {"range", []interface{}{keyBucketName, newTestRevBytes(revision{1, 0}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(restoreChunkKeys)}},
  355. }
  356. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  357. t.Errorf("tx actions = %+v, want %+v", g, wact)
  358. }
  359. gens := []generation{
  360. {created: revision{4, 0}, ver: 2, revs: []revision{{3, 0}, {5, 0}}},
  361. {created: revision{0, 0}, ver: 0, revs: nil},
  362. }
  363. ki := &keyIndex{key: []byte("foo"), modified: revision{5, 0}, generations: gens}
  364. wact = []testutil.Action{
  365. {"keyIndex", []interface{}{ki}},
  366. {"insert", []interface{}{ki}},
  367. }
  368. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  369. t.Errorf("index action = %+v, want %+v", g, wact)
  370. }
  371. }
  372. func TestRestoreDelete(t *testing.T) {
  373. oldChunk := restoreChunkKeys
  374. restoreChunkKeys = mrand.Intn(3) + 2
  375. defer func() { restoreChunkKeys = oldChunk }()
  376. b, tmpPath := backend.NewDefaultTmpBackend()
  377. s := NewStore(b, &lease.FakeLessor{}, nil)
  378. defer os.Remove(tmpPath)
  379. keys := make(map[string]struct{})
  380. for i := 0; i < 20; i++ {
  381. ks := fmt.Sprintf("foo-%d", i)
  382. k := []byte(ks)
  383. s.Put(k, []byte("bar"), lease.NoLease)
  384. keys[ks] = struct{}{}
  385. switch mrand.Intn(3) {
  386. case 0:
  387. // put random key from past via random range on map
  388. ks = fmt.Sprintf("foo-%d", mrand.Intn(i+1))
  389. s.Put([]byte(ks), []byte("baz"), lease.NoLease)
  390. keys[ks] = struct{}{}
  391. case 1:
  392. // delete random key via random range on map
  393. for k := range keys {
  394. s.DeleteRange([]byte(k), nil)
  395. delete(keys, k)
  396. break
  397. }
  398. }
  399. }
  400. s.Close()
  401. s = NewStore(b, &lease.FakeLessor{}, nil)
  402. defer s.Close()
  403. for i := 0; i < 20; i++ {
  404. ks := fmt.Sprintf("foo-%d", i)
  405. r, err := s.Range([]byte(ks), nil, RangeOptions{})
  406. if err != nil {
  407. t.Fatal(err)
  408. }
  409. if _, ok := keys[ks]; ok {
  410. if len(r.KVs) == 0 {
  411. t.Errorf("#%d: expected %q, got deleted", i, ks)
  412. }
  413. } else if len(r.KVs) != 0 {
  414. t.Errorf("#%d: expected deleted, got %q", i, ks)
  415. }
  416. }
  417. }
  418. func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
  419. b, tmpPath := backend.NewDefaultTmpBackend()
  420. s0 := NewStore(b, &lease.FakeLessor{}, nil)
  421. defer os.Remove(tmpPath)
  422. s0.Put([]byte("foo"), []byte("bar"), lease.NoLease)
  423. s0.Put([]byte("foo"), []byte("bar1"), lease.NoLease)
  424. s0.Put([]byte("foo"), []byte("bar2"), lease.NoLease)
  425. // write scheduled compaction, but not do compaction
  426. rbytes := newRevBytes()
  427. revToBytes(revision{main: 2}, rbytes)
  428. tx := s0.b.BatchTx()
  429. tx.Lock()
  430. tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
  431. tx.Unlock()
  432. s0.Close()
  433. s1 := NewStore(b, &lease.FakeLessor{}, nil)
  434. // wait for scheduled compaction to be finished
  435. time.Sleep(100 * time.Millisecond)
  436. if _, err := s1.Range([]byte("foo"), nil, RangeOptions{Rev: 1}); err != ErrCompacted {
  437. t.Errorf("range on compacted rev error = %v, want %v", err, ErrCompacted)
  438. }
  439. // check the key in backend is deleted
  440. revbytes := newRevBytes()
  441. revToBytes(revision{main: 1}, revbytes)
  442. // The disk compaction is done asynchronously and requires more time on slow disk.
  443. // try 5 times for CI with slow IO.
  444. for i := 0; i < 5; i++ {
  445. tx = s1.b.BatchTx()
  446. tx.Lock()
  447. ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
  448. tx.Unlock()
  449. if len(ks) != 0 {
  450. time.Sleep(100 * time.Millisecond)
  451. continue
  452. }
  453. return
  454. }
  455. t.Errorf("key for rev %+v still exists, want deleted", bytesToRev(revbytes))
  456. }
  457. func TestTxnPut(t *testing.T) {
  458. // assign arbitrary size
  459. bytesN := 30
  460. sliceN := 100
  461. keys := createBytesSlice(bytesN, sliceN)
  462. vals := createBytesSlice(bytesN, sliceN)
  463. b, tmpPath := backend.NewDefaultTmpBackend()
  464. s := NewStore(b, &lease.FakeLessor{}, nil)
  465. defer cleanup(s, b, tmpPath)
  466. for i := 0; i < sliceN; i++ {
  467. txn := s.Write()
  468. base := int64(i + 2)
  469. if rev := txn.Put(keys[i], vals[i], lease.NoLease); rev != base {
  470. t.Errorf("#%d: rev = %d, want %d", i, rev, base)
  471. }
  472. txn.End()
  473. }
  474. }
  475. func TestTxnBlockBackendForceCommit(t *testing.T) {
  476. b, tmpPath := backend.NewDefaultTmpBackend()
  477. s := NewStore(b, &lease.FakeLessor{}, nil)
  478. defer os.Remove(tmpPath)
  479. txn := s.Read()
  480. done := make(chan struct{})
  481. go func() {
  482. s.b.ForceCommit()
  483. done <- struct{}{}
  484. }()
  485. select {
  486. case <-done:
  487. t.Fatalf("failed to block ForceCommit")
  488. case <-time.After(100 * time.Millisecond):
  489. }
  490. txn.End()
  491. select {
  492. case <-done:
  493. case <-time.After(5 * time.Second): // wait 5 seconds for CI with slow IO
  494. testutil.FatalStack(t, "failed to execute ForceCommit")
  495. }
  496. }
  497. // TODO: test attach key to lessor
  498. func newTestRevBytes(rev revision) []byte {
  499. bytes := newRevBytes()
  500. revToBytes(rev, bytes)
  501. return bytes
  502. }
  503. func newTestKeyBytes(rev revision, tombstone bool) []byte {
  504. bytes := newRevBytes()
  505. revToBytes(rev, bytes)
  506. if tombstone {
  507. bytes = appendMarkTombstone(bytes)
  508. }
  509. return bytes
  510. }
  511. func newFakeStore() *store {
  512. b := &fakeBackend{&fakeBatchTx{
  513. Recorder: &testutil.RecorderBuffered{},
  514. rangeRespc: make(chan rangeResp, 5)}}
  515. fi := &fakeIndex{
  516. Recorder: &testutil.RecorderBuffered{},
  517. indexGetRespc: make(chan indexGetResp, 1),
  518. indexRangeRespc: make(chan indexRangeResp, 1),
  519. indexRangeEventsRespc: make(chan indexRangeEventsResp, 1),
  520. indexCompactRespc: make(chan map[revision]struct{}, 1),
  521. }
  522. s := &store{
  523. b: b,
  524. le: &lease.FakeLessor{},
  525. kvindex: fi,
  526. currentRev: 0,
  527. compactMainRev: -1,
  528. fifoSched: schedule.NewFIFOScheduler(),
  529. stopc: make(chan struct{}),
  530. }
  531. s.ReadView, s.WriteView = &readView{s}, &writeView{s}
  532. return s
  533. }
  534. type rangeResp struct {
  535. keys [][]byte
  536. vals [][]byte
  537. }
  538. type fakeBatchTx struct {
  539. testutil.Recorder
  540. rangeRespc chan rangeResp
  541. }
  542. func (b *fakeBatchTx) Lock() {}
  543. func (b *fakeBatchTx) Unlock() {}
  544. func (b *fakeBatchTx) UnsafeCreateBucket(name []byte) {}
  545. func (b *fakeBatchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
  546. b.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{bucketName, key, value}})
  547. }
  548. func (b *fakeBatchTx) UnsafeSeqPut(bucketName []byte, key []byte, value []byte) {
  549. b.Recorder.Record(testutil.Action{Name: "seqput", Params: []interface{}{bucketName, key, value}})
  550. }
  551. func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
  552. b.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{bucketName, key, endKey, limit}})
  553. r := <-b.rangeRespc
  554. return r.keys, r.vals
  555. }
  556. func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
  557. b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
  558. }
  559. func (b *fakeBatchTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
  560. return nil
  561. }
  562. func (b *fakeBatchTx) Commit() {}
  563. func (b *fakeBatchTx) CommitAndStop() {}
  564. type fakeBackend struct {
  565. tx *fakeBatchTx
  566. }
  567. func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
  568. func (b *fakeBackend) ReadTx() backend.ReadTx { return b.tx }
  569. func (b *fakeBackend) Hash(ignores map[backend.IgnoreKey]struct{}) (uint32, error) { return 0, nil }
  570. func (b *fakeBackend) Size() 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. }