mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-17 17:33:47 +00:00
A workaround for LocalStore put corrupting the chunk key
This commit is contained in:
parent
6bf67decd2
commit
f43390f5af
4 changed files with 42 additions and 9 deletions
|
|
@ -17,6 +17,7 @@
|
||||||
package stream
|
package stream
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
|
|
@ -197,6 +198,9 @@ func (d *Delivery) processReceivedChunks() {
|
||||||
// this should be has locally
|
// this should be has locally
|
||||||
log.Error("before db.Get", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex())
|
log.Error("before db.Get", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex())
|
||||||
chunk, err := d.db.Get(req.Key)
|
chunk, err := d.db.Get(req.Key)
|
||||||
|
if !bytes.Equal(chunk.Key, req.Key) {
|
||||||
|
panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID()))
|
||||||
|
}
|
||||||
log.Error("after db.Get", "peer", req.peer.ID(), "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
|
log.Error("after db.Get", "peer", req.peer.ID(), "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
log.Error("found existing?", "peer", req.peer.ID(), "hash", chunk.Key.Hex())
|
log.Error("found existing?", "peer", req.peer.ID(), "hash", chunk.Key.Hex())
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,7 @@ import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
|
|
@ -34,7 +35,6 @@ import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
"github.com/ethereum/go-ethereum/rlp"
|
|
||||||
"github.com/ethereum/go-ethereum/swarm/storage/mock"
|
"github.com/ethereum/go-ethereum/swarm/storage/mock"
|
||||||
"github.com/syndtr/goleveldb/leveldb"
|
"github.com/syndtr/goleveldb/leveldb"
|
||||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||||
|
|
@ -221,7 +221,12 @@ func getDataKey(idx uint64, po uint8) []byte {
|
||||||
}
|
}
|
||||||
|
|
||||||
func encodeIndex(index *dpaDBIndex) []byte {
|
func encodeIndex(index *dpaDBIndex) []byte {
|
||||||
data, _ := rlp.EncodeToBytes(index)
|
//data, _ := rlp.EncodeToBytes(index)
|
||||||
|
|
||||||
|
data, err := json.Marshal(index)
|
||||||
|
if err != nil {
|
||||||
|
panic(err)
|
||||||
|
}
|
||||||
return data
|
return data
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -230,8 +235,10 @@ func encodeData(chunk *Chunk) []byte {
|
||||||
}
|
}
|
||||||
|
|
||||||
func decodeIndex(data []byte, index *dpaDBIndex) error {
|
func decodeIndex(data []byte, index *dpaDBIndex) error {
|
||||||
dec := rlp.NewStream(bytes.NewReader(data), 0)
|
// dec := rlp.NewStream(bytes.NewReader(data), 0)
|
||||||
return dec.Decode(index)
|
// return dec.Decode(index)
|
||||||
|
return json.Unmarshal(data, index)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func decodeData(data []byte, chunk *Chunk) {
|
func decodeData(data []byte, chunk *Chunk) {
|
||||||
|
|
@ -542,6 +549,7 @@ func (s *DbStore) Put(chunk *Chunk) {
|
||||||
log.Error("DbStore.Put", "hash", chunk.Key.Hex())
|
log.Error("DbStore.Put", "hash", chunk.Key.Hex())
|
||||||
done := make(chan struct{})
|
done := make(chan struct{})
|
||||||
defer close(done)
|
defer close(done)
|
||||||
|
key := Key(append(make([]byte, 0), chunk.Key...))
|
||||||
go func() {
|
go func() {
|
||||||
log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex())
|
log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex())
|
||||||
select {
|
select {
|
||||||
|
|
@ -549,6 +557,9 @@ func (s *DbStore) Put(chunk *Chunk) {
|
||||||
log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex())
|
log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex())
|
||||||
case <-done:
|
case <-done:
|
||||||
log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex())
|
log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex())
|
||||||
|
if !bytes.Equal(chunk.Key, key) {
|
||||||
|
panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
|
@ -724,6 +735,10 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) {
|
||||||
chunk = NewChunk(key, nil)
|
chunk = NewChunk(key, nil)
|
||||||
decodeData(data, chunk)
|
decodeData(data, chunk)
|
||||||
|
|
||||||
|
if !bytes.Equal(chunk.Key, key) {
|
||||||
|
panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex()))
|
||||||
|
}
|
||||||
|
|
||||||
} else {
|
} else {
|
||||||
err = ErrNotFound
|
err = ErrNotFound
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@
|
||||||
package storage
|
package storage
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"fmt"
|
"fmt"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
|
@ -95,10 +96,18 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error)
|
||||||
// unsafe, in that the data is not integrity checked
|
// unsafe, in that the data is not integrity checked
|
||||||
func (self *LocalStore) Put(chunk *Chunk) {
|
func (self *LocalStore) Put(chunk *Chunk) {
|
||||||
chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
|
chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
|
||||||
self.memStore.Put(chunk)
|
c := &Chunk{
|
||||||
log.Error("put to memstore", "hash", chunk.Key.Hex())
|
Key: Key(append([]byte{}, chunk.Key...)),
|
||||||
self.DbStore.Put(chunk)
|
SData: append([]byte{}, chunk.SData...),
|
||||||
log.Error("put to dbstore", "hash", chunk.Key.Hex())
|
dbStored: chunk.dbStored,
|
||||||
|
}
|
||||||
|
self.memStore.Put(c)
|
||||||
|
log.Error("put to memstore", "hash", c.Key.Hex())
|
||||||
|
self.DbStore.Put(c)
|
||||||
|
log.Error("put to dbstore", "hash", c.Key.Hex())
|
||||||
|
if !bytes.Equal(chunk.Key, c.Key) {
|
||||||
|
panic(fmt.Errorf("LocalStore.Put: chunk %s != c %s", chunk.Key.Hex(), c.Key.Hex()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get(chunk *Chunk) looks up a chunk in the local stores
|
// Get(chunk *Chunk) looks up a chunk in the local stores
|
||||||
|
|
@ -122,7 +131,7 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
|
chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
|
||||||
self.memStore.Put(chunk)
|
//self.memStore.Put(chunk)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,8 @@
|
||||||
package storage
|
package storage
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
|
"fmt"
|
||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -329,6 +331,9 @@ func (m *MemStore) Get(key Key) (*Chunk, error) {
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, ErrNotFound
|
return nil, ErrNotFound
|
||||||
}
|
}
|
||||||
|
if !bytes.Equal(c.Key, key) {
|
||||||
|
panic(fmt.Errorf("MemStore.Get: chunk key %s != req key %s", c.Key.Hex(), key.Hex()))
|
||||||
|
}
|
||||||
return c, nil
|
return c, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue