forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-31 09:03:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat: refactor storage interface

Browse source
This commit is contained in:
fearlessfe 2024-03-22 22:40:47 +08:00 committed by Chen Kai
parent e3c91b3fc5
commit 6e23955304
8 changed files with 74 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
p2p/discover/api.go
View file

@ -476,7 +476,7 @@ func (p *PortalProtocolAPI) LocalContent(contentKeyHex string) (string, error) {
return "", err
}
contentId := p.portalProtocol.ToContentId(contentKey)
content, err := p.portalProtocol.Get(contentId)
content, err := p.portalProtocol.Get(contentKey, contentId)
if err != nil {
return "", err
@ -497,7 +497,7 @@ func (p *PortalProtocolAPI) Store(contentKeyHex string, contextHex string) (bool
if err != nil {
return false, err
}
err = p.portalProtocol.Put(contentId, content)
err = p.portalProtocol.Put(contentKey, contentId, content)
if err != nil {
return false, err
}

24
p2p/discover/portal_protocol.go
View file

@ -531,7 +531,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
contentKey := request.Request.(*PersistOfferRequest).ContentKeys[index]
contentId := p.toContentId(contentKey)
if contentId != nil {
content, err = p.storage.Get(contentId)
content, err = p.storage.Get(contentKey, contentId)
if err != nil {
p.log.Error("failed to get content from storage", "err", err)
contents = append(contents, []byte{})
@ -910,14 +910,14 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
maxPayloadSize := maxPacketSize - talkRespOverhead - contentOverhead
enrOverhead := 4 //per added ENR, 4 bytes offset overhead
var err error
contentId := p.toContentId(request.ContentKey)
contentKey := request.ContentKey
contentId := p.toContentId(contentKey)
if contentId == nil {
return nil, ErrNilContentKey
}
var content []byte
content, err = p.storage.Get(contentId)
content, err = p.storage.Get(contentKey, contentId)
if err != nil && !errors.Is(err, ContentNotFound) {
return nil, err
}
@ -1088,7 +1088,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
contentId := p.toContentId(contentKey)
if contentId != nil {
if inRange(p.Self().ID(), p.nodeRadius, contentId) {
if _, err = p.storage.Get(contentId); err != nil {
if _, err = p.storage.Get(contentKey, contentId); err != nil {
contentKeyBitlist.SetBitAt(uint64(i), true)
contentKeys = append(contentKeys, contentKey)
}
@ -1426,7 +1426,9 @@ func (p *PortalProtocol) ContentLookup(contentKey []byte) ([]byte, bool, error)
defer cancel()
resChan := make(chan *ContentInfoResp, 1)
defer close(resChan)
newLookup(lookupContext, p.table, p.Self().ID(), func(n *node) ([]*node, error) {
contentId := p.ToContentId(contentKey)
newLookup(lookupContext, p.table, enode.ID(contentId), func(n *node) ([]*node, error) {
return p.contentLookupWorker(unwrapNode(n), contentKey, resChan)
}).run()
@ -1528,7 +1530,7 @@ func (p *PortalProtocol) TraceContentLookup(contentKey []byte) (*TraceContentRes
}
}()
newLookup(lookupContext, p.table, p.Self().ID(), func(n *node) ([]*node, error) {
newLookup(lookupContext, p.table, enode.ID(contentId), func(n *node) ([]*node, error) {
node := unwrapNode(n)
requestNodeChan <- node
return p.traceContentLookupWorker(node, contentKey, resChan)
@ -1618,14 +1620,14 @@ func (p *PortalProtocol) InRange(contentId []byte) bool {
return inRange(p.Self().ID(), p.nodeRadius, contentId)
}
func (p *PortalProtocol) Get(contentId []byte) ([]byte, error) {
content, err := p.storage.Get(contentId)
func (p *PortalProtocol) Get(contentKey []byte, contentId []byte) ([]byte, error) {
content, err := p.storage.Get(contentKey, contentId)
p.log.Trace("get local storage", "contentId", hexutil.Encode(contentId), "content", hexutil.Encode(content), "err", err)
return content, err
}
func (p *PortalProtocol) Put(contentId []byte, content []byte) error {
err := p.storage.Put(contentId, content)
func (p *PortalProtocol) Put(contentKey []byte, contentId []byte, content []byte) error {
err := p.storage.Put(contentKey, contentId, content)
p.log.Trace("put local storage", "contentId", hexutil.Encode(contentId), "content", hexutil.Encode(content), "err", err)
return err
}

12
p2p/discover/portal_protocol_test.go
View file

@ -27,14 +27,14 @@ type MockStorage struct {
db map[string][]byte
}
func (m *MockStorage) Get(contentId []byte) ([]byte, error) {
func (m *MockStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
if content, ok := m.db[string(contentId)]; ok {
return content, nil
}
return nil, ContentNotFound
}
func (m *MockStorage) Put(contentId []byte, content []byte) error {
func (m *MockStorage) Put(contentKey []byte, contentId []byte, content []byte) error {
m.db[string(contentId)] = content
return nil
}
@ -304,7 +304,7 @@ func TestPortalWireProtocol(t *testing.T) {
return n.ID() == node2.localNode.Node().ID()
})
err = node1.storage.Put(node1.toContentId([]byte("test_key")), []byte("test_value"))
err = node1.storage.Put(nil, node1.toContentId([]byte("test_key")), []byte("test_value"))
assert.NoError(t, err)
flag, content, err := node2.findContent(node1.localNode.Node(), []byte("test_key"))
@ -324,7 +324,7 @@ func TestPortalWireProtocol(t *testing.T) {
_, err = rand.Read(largeTestContent)
assert.NoError(t, err)
err = node1.storage.Put(node1.toContentId([]byte("large_test_key")), largeTestContent)
err = node1.storage.Put(nil, node1.toContentId([]byte("large_test_key")), largeTestContent)
assert.NoError(t, err)
flag, content, err = node2.findContent(node1.localNode.Node(), []byte("large_test_key"))
@ -452,7 +452,7 @@ func TestContentLookup(t *testing.T) {
content := []byte{0x1, 0x2}
contentId := node1.toContentId(contentKey)
err = node3.storage.Put(contentId, content)
err = node3.storage.Put(nil, contentId, content)
assert.NoError(t, err)
res, _, err := node1.ContentLookup(contentKey)
@ -487,7 +487,7 @@ func TestTraceContentLookup(t *testing.T) {
content := []byte{0x1, 0x2}
contentId := node1.toContentId(contentKey)
err = node1.storage.Put(contentId, content)
err = node1.storage.Put(nil, contentId, content)
assert.NoError(t, err)
node1Id := hexutil.Encode(node1.Self().ID().Bytes())

48
portalnetwork/history/history_network.go
View file

@ -102,7 +102,7 @@ func (h *HistoryNetwork) GetBlockHeader(blockHash []byte) (*types.Header, error)
return nil, ErrContentOutOfRange
}
res, err := h.portalProtocol.Get(contentId)
res, err := h.portalProtocol.Get(contentKey, contentId)
// other error
if err != nil && !errors.Is(err, storage.ErrContentNotFound) {
return nil, err
@ -119,27 +119,32 @@ func (h *HistoryNetwork) GetBlockHeader(blockHash []byte) (*types.Header, error)
}
// no content in local storage
for retries := 0; retries < requestRetries; retries++ {
// TODO log the err and continue
content, _, err := h.portalProtocol.ContentLookup(contentKey)
if err != nil {
h.log.Error("getBlockHeader failed", "contentKey", hexutil.Encode(contentKey), "err", err)
continue
}
headerWithProof, err := DecodeBlockHeaderWithProof(content)
if err != nil {
h.log.Error("decodeBlockHeaderWithProof failed", "content", hexutil.Encode(content), "err", err)
continue
}
header, err := ValidateBlockHeaderBytes(headerWithProof.Header, blockHash)
if err != nil {
h.log.Error("validateBlockHeaderBytes failed", "header", hexutil.Encode(headerWithProof.Header), "blockhash", hexutil.Encode(blockHash), "err", err)
continue
}
valid, err := h.verifyHeader(header, *headerWithProof.Proof)
if err != nil || !valid {
h.log.Error("verifyHeader failed", "err", err)
continue
}
// TODO handle the error
_ = h.portalProtocol.Put(contentId, content)
err = h.portalProtocol.Put(contentKey, contentId, content)
if err != nil {
h.log.Error("failed to store content in getBlockHeader", "contentKey", hexutil.Encode(contentKey), "content", hexutil.Encode(content))
}
return header, nil
}
return nil, storage.ErrContentNotFound
@ -157,7 +162,7 @@ func (h *HistoryNetwork) GetBlockBody(blockHash []byte) (*types.Body, error) {
return nil, ErrContentOutOfRange
}
res, err := h.portalProtocol.Get(contentId)
res, err := h.portalProtocol.Get(contentKey, contentId)
// other error
// TODO maybe use nil res to replace the ErrContentNotFound
if err != nil && err != storage.ErrContentNotFound {
@ -173,19 +178,24 @@ func (h *HistoryNetwork) GetBlockBody(blockHash []byte) (*types.Body, error) {
for retries := 0; retries < requestRetries; retries++ {
content, _, err := h.portalProtocol.ContentLookup(contentKey)
if err != nil {
h.log.Error("getBlockBody failed", "contentKey", hexutil.Encode(contentKey), "err", err)
continue
}
body, err := DecodePortalBlockBodyBytes(content)
if err != nil {
h.log.Error("decodePortalBlockBodyBytes failed", "content", hexutil.Encode(content), "err", err)
continue
}
err = validateBlockBody(body, header)
if err != nil {
h.log.Error("validateBlockBody failed", "header", "err", err)
continue
}
// TODO handle the error
_ = h.portalProtocol.Put(contentId, content)
err = h.portalProtocol.Put(contentKey, contentId, content)
if err != nil {
h.log.Error("failed to store content in getBlockBody", "contentKey", hexutil.Encode(contentKey), "content", hexutil.Encode(content))
}
return body, nil
}
return nil, storage.ErrContentNotFound
@ -203,7 +213,7 @@ func (h *HistoryNetwork) GetReceipts(blockHash []byte) ([]*types.Receipt, error)
return nil, ErrContentOutOfRange
}
res, err := h.portalProtocol.Get(contentId)
res, err := h.portalProtocol.Get(contentKey, contentId)
// other error
if err != nil && err != storage.ErrContentNotFound {
return nil, err
@ -223,14 +233,18 @@ func (h *HistoryNetwork) GetReceipts(blockHash []byte) ([]*types.Receipt, error)
for retries := 0; retries < requestRetries; retries++ {
content, _, err := h.portalProtocol.ContentLookup(contentKey)
if err != nil {
h.log.Error("getReceipts failed", "contentKey", hexutil.Encode(contentKey), "err", err)
continue
}
receipts, err := ValidatePortalReceiptsBytes(content, header.ReceiptHash.Bytes())
if err != nil {
h.log.Error("getReceipts failed", "err", err)
continue
}
// TODO handle the error
_ = h.portalProtocol.Put(contentId, content)
err = h.portalProtocol.Put(contentKey, contentId, content)
if err != nil {
h.log.Error("failed to store content in getReceipts", "contentKey", hexutil.Encode(contentKey), "content", hexutil.Encode(content))
}
return receipts, nil
}
return nil, storage.ErrContentNotFound
@ -240,7 +254,7 @@ func (h *HistoryNetwork) GetEpochAccumulator(epochHash []byte) (*EpochAccumulato
contentKey := newContentKey(EpochAccumulatorType, epochHash).encode()
contentId := h.portalProtocol.ToContentId(contentKey)
res, err := h.portalProtocol.Get(contentId)
res, err := h.portalProtocol.Get(contentKey, contentId)
// other error
if err != nil && err != storage.ErrContentNotFound {
return nil, err
@ -253,22 +267,28 @@ func (h *HistoryNetwork) GetEpochAccumulator(epochHash []byte) (*EpochAccumulato
for retries := 0; retries < requestRetries; retries++ {
content, _, err := h.portalProtocol.ContentLookup(contentKey)
if err != nil {
h.log.Error("getEpochAccumulator failed", "contentKey", hexutil.Encode(contentKey), "err", err)
continue
}
epochAccu, err := decodeEpochAccumulator(content)
if err != nil {
h.log.Error("decodeEpochAccumulator failed", "content", hexutil.Encode(content), "err", err)
continue
}
hash, err := epochAccu.HashTreeRoot()
if err != nil {
h.log.Error("hashTreeRoot failed", "err", err)
continue
}
mixHash := MixInLength(hash, epochSize)
if !bytes.Equal(mixHash, epochHash) {
h.log.Error("epochHash is not equal", "mixHash", hexutil.Encode(mixHash), "epochHash", hexutil.Encode(epochHash))
continue
}
// TODO handle the error
_ = h.portalProtocol.Put(contentId, content)
err = h.portalProtocol.Put(contentKey, contentId, content)
if err != nil {
h.log.Error("failed to store content in getReceipts", "contentKey", hexutil.Encode(contentKey), "content", hexutil.Encode(content))
}
return epochAccu, nil
}
return nil, storage.ErrContentNotFound
@ -569,7 +589,7 @@ func (h *HistoryNetwork) validateContents(contentKeys [][]byte, contents [][]byt
return fmt.Errorf("content validate failed with content key %x and content %x", contentKey, content)
}
contentId := h.portalProtocol.ToContentId(contentKey)
_ = h.portalProtocol.Put(contentId, content)
_ = h.portalProtocol.Put(contentKey, contentId, content)
}
return nil
}

12
portalnetwork/history/history_network_test.go
View file

@ -206,7 +206,7 @@ func TestGetContentByKey(t *testing.T) {
require.Nil(t, header)
contentId := historyNetwork1.portalProtocol.ToContentId(headerEntry.key)
err = historyNetwork1.portalProtocol.Put(contentId, headerEntry.value)
err = historyNetwork1.portalProtocol.Put(headerEntry.key, contentId, headerEntry.value)
require.NoError(t, err)
// get content from historyNetwork1
header, err = historyNetwork2.GetBlockHeader(headerEntry.key[1:])
@ -225,7 +225,7 @@ func TestGetContentByKey(t *testing.T) {
require.Nil(t, body)
contentId = historyNetwork1.portalProtocol.ToContentId(bodyEntry.key)
err = historyNetwork1.portalProtocol.Put(contentId, bodyEntry.value)
err = historyNetwork1.portalProtocol.Put(bodyEntry.key, contentId, bodyEntry.value)
require.NoError(t, err)
// get content from historyNetwork1
body, err = historyNetwork2.GetBlockBody(bodyEntry.key[1:])
@ -244,7 +244,7 @@ func TestGetContentByKey(t *testing.T) {
require.Nil(t, receipts)
contentId = historyNetwork1.portalProtocol.ToContentId(receiptsEntry.key)
err = historyNetwork1.portalProtocol.Put(contentId, receiptsEntry.value)
err = historyNetwork1.portalProtocol.Put(receiptsEntry.key, contentId, receiptsEntry.value)
require.NoError(t, err)
// get content from historyNetwork1
receipts, err = historyNetwork2.GetReceipts(receiptsEntry.key[1:])
@ -276,7 +276,7 @@ func TestGetContentByKey(t *testing.T) {
require.Nil(t, epoch)
contentId = historyNetwork1.portalProtocol.ToContentId(contentKey)
err = historyNetwork1.portalProtocol.Put(contentId, content)
err = historyNetwork1.portalProtocol.Put(contentKey, contentId, content)
require.NoError(t, err)
// get content from historyNetwork1
epoch, err = historyNetwork2.GetEpochAccumulator(contentKey[1:])
@ -366,14 +366,14 @@ type MockStorage struct {
db map[string][]byte
}
func (m *MockStorage) Get(contentId []byte) ([]byte, error) {
func (m *MockStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
if content, ok := m.db[string(contentId)]; ok {
return content, nil
}
return nil, storage.ErrContentNotFound
}
func (m *MockStorage) Put(contentId []byte, content []byte) error {
func (m *MockStorage) Put(contentKey []byte, contentId []byte, content []byte) error {
m.db[string(contentId)] = content
return nil
}

4
portalnetwork/storage/content_storage.go
View file

@ -5,7 +5,7 @@ import "fmt"
var ErrContentNotFound = fmt.Errorf("content not found")
type ContentStorage interface {
Get(contentId []byte) ([]byte, error)
Get(contentKey []byte, contentId []byte) ([]byte, error)
Put(contentId []byte, content []byte) error
Put(contentKey []byte, contentId []byte, content []byte) error
}

4
portalnetwork/storage/sqlite/content_storage.go
View file

@ -145,7 +145,7 @@ func createDir(dir string) error {
}
// Get the content according to the contentId
func (p *ContentStorage) Get(contentId []byte) ([]byte, error) {
func (p *ContentStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
var res []byte
err := p.getStmt.QueryRow(contentId).Scan(&res)
if err == sql.ErrNoRows {
@ -178,7 +178,7 @@ func newPutResultWithErr(err error) PutResult {
}
}
func (p *ContentStorage) Put(contentId []byte, content []byte) error {
func (p *ContentStorage) Put(contentKey []byte, contentId []byte, content []byte) error {
res := p.put(contentId, content)
return res.Err()
}

18
portalnetwork/storage/sqlite/content_storage_test.go
View file

@ -36,13 +36,13 @@ func TestBasicStorage(t *testing.T) {
contentId := []byte("test")
content := []byte("value")
_, err = storage.Get(contentId)
_, err = storage.Get(nil, contentId)
assert.Equal(t, contentStorage.ErrContentNotFound, err)
pt := storage.put(contentId, content)
assert.NoError(t, pt.Err())
val, err := storage.Get(contentId)
val, err := storage.Get(nil, contentId)
assert.NoError(t, err)
assert.Equal(t, content, val)
@ -177,13 +177,13 @@ func TestDBPruning(t *testing.T) {
assert.NoError(t, err)
assert.True(t, usedSize < storage.storageCapacityInBytes)
_, err = storage.Get(furthestElement.Bytes())
_, err = storage.Get(nil, furthestElement.Bytes())
assert.Equal(t, contentStorage.ErrContentNotFound, err)
_, err = storage.Get(secondFurthest.Bytes())
_, err = storage.Get(nil, secondFurthest.Bytes())
assert.Equal(t, contentStorage.ErrContentNotFound, err)
val, err := storage.Get(thirdFurthest.Bytes())
val, err := storage.Get(nil, thirdFurthest.Bytes())
assert.NoError(t, err)
assert.NotNil(t, val)
}
@ -203,7 +203,7 @@ func TestGetLargestDistance(t *testing.T) {
pt7 := storage.put(furthestElement.Bytes(), genBytes(2000))
assert.NoError(t, pt7.Err())
val, err := storage.Get(furthestElement.Bytes())
val, err := storage.Get(nil, furthestElement.Bytes())
assert.NoError(t, err)
assert.NotNil(t, val)
pt8 := storage.put(secondFurthest.Bytes(), genBytes(2000))
@ -241,13 +241,13 @@ func TestSimpleForcePruning(t *testing.T) {
err = storage.ForcePrune(uint256.NewInt(20))
assert.NoError(t, err)
_, err = storage.Get(furthestElement.Bytes())
_, err = storage.Get(nil, furthestElement.Bytes())
assert.Equal(t, contentStorage.ErrContentNotFound, err)
_, err = storage.Get(secondFurthest.Bytes())
_, err = storage.Get(nil, secondFurthest.Bytes())
assert.Equal(t, contentStorage.ErrContentNotFound, err)
_, err = storage.Get(third.Bytes())
_, err = storage.Get(nil, third.Bytes())
assert.NoError(t, err)
}