[#1329] services/tree: Implement GetOpLog
RPC
Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
parent
886baf3136
commit
536857ea5a
11 changed files with 195 additions and 3 deletions
|
@ -116,3 +116,22 @@ func (e *StorageEngine) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pil
|
|||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// TreeGetOpLog implements the pilorama.Forest interface.
|
||||
func (e *StorageEngine) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
|
||||
var err error
|
||||
var lm pilorama.Move
|
||||
for _, sh := range e.sortShardsByWeight(cid) {
|
||||
lm, err = sh.TreeGetOpLog(cid, treeID, height)
|
||||
if err != nil {
|
||||
e.log.Debug("can't perform `GetOpLog`",
|
||||
zap.Stringer("cid", cid),
|
||||
zap.String("tree", treeID),
|
||||
zap.Uint64("height", height),
|
||||
zap.String("err", err.Error()))
|
||||
continue
|
||||
}
|
||||
return lm, nil
|
||||
}
|
||||
return lm, err
|
||||
}
|
||||
|
|
|
@ -211,7 +211,7 @@ func (t *boltForest) applyOperation(logBucket, treeBucket *bbolt.Bucket, m *Move
|
|||
|
||||
// 1. Undo up until the desired timestamp is here.
|
||||
for len(key) == 8 && binary.BigEndian.Uint64(key) > m.Time {
|
||||
if err := t.logFromBytes(&tmp, key, value); err != nil {
|
||||
if err := t.logFromBytes(&tmp, value); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := t.undo(&tmp.Move, &tmp, treeBucket, cKey[:]); err != nil {
|
||||
|
@ -231,7 +231,7 @@ func (t *boltForest) applyOperation(logBucket, treeBucket *bbolt.Bucket, m *Move
|
|||
|
||||
// 3. Re-apply all other operations.
|
||||
for len(key) == 8 {
|
||||
if err := t.logFromBytes(&tmp, key, value); err != nil {
|
||||
if err := t.logFromBytes(&tmp, value); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := t.do(logBucket, treeBucket, cKey[:], &tmp); err != nil {
|
||||
|
@ -446,6 +446,29 @@ func (t *boltForest) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node)
|
|||
return children, err
|
||||
}
|
||||
|
||||
// TreeGetOpLog implements the pilorama.Forest interface.
|
||||
func (t *boltForest) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error) {
|
||||
key := make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(key, height)
|
||||
|
||||
var lm Move
|
||||
|
||||
err := t.db.View(func(tx *bbolt.Tx) error {
|
||||
treeRoot := tx.Bucket(bucketName(cid, treeID))
|
||||
if treeRoot == nil {
|
||||
return ErrTreeNotFound
|
||||
}
|
||||
|
||||
c := treeRoot.Bucket(logBucket).Cursor()
|
||||
if _, data := c.Seek(key); data != nil {
|
||||
return t.moveFromBytes(&lm, data)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
return lm, err
|
||||
}
|
||||
|
||||
func (t *boltForest) getPathPrefix(bTree *bbolt.Bucket, attr string, path []string) (int, Node, error) {
|
||||
var key [9]byte
|
||||
|
||||
|
@ -483,7 +506,17 @@ loop:
|
|||
return len(path), curNode, nil
|
||||
}
|
||||
|
||||
func (t *boltForest) logFromBytes(lm *LogMove, key []byte, data []byte) error {
|
||||
func (t *boltForest) moveFromBytes(m *Move, data []byte) error {
|
||||
r := io.NewBinReaderFromBuf(data)
|
||||
m.Child = r.ReadU64LE()
|
||||
m.Parent = r.ReadU64LE()
|
||||
if err := m.Meta.FromBytes(r.ReadVarBytes()); err != nil {
|
||||
return err
|
||||
}
|
||||
return r.Err
|
||||
}
|
||||
|
||||
func (t *boltForest) logFromBytes(lm *LogMove, data []byte) error {
|
||||
r := io.NewBinReaderFromBuf(data)
|
||||
lm.Child = r.ReadU64LE()
|
||||
lm.Parent = r.ReadU64LE()
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
package pilorama
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
cidSDK "github.com/nspcc-dev/neofs-sdk-go/container/id"
|
||||
)
|
||||
|
||||
|
@ -133,3 +135,20 @@ func (f *memoryForest) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node
|
|||
copy(res, children)
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// TreeGetOpLog implements the pilorama.Forest interface.
|
||||
func (f *memoryForest) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error) {
|
||||
fullID := cid.String() + "/" + treeID
|
||||
s, ok := f.treeMap[fullID]
|
||||
if !ok {
|
||||
return Move{}, ErrTreeNotFound
|
||||
}
|
||||
|
||||
n := sort.Search(len(s.operations), func(i int) bool {
|
||||
return s.operations[i].Time >= height
|
||||
})
|
||||
if n == len(s.operations) {
|
||||
return Move{}, nil
|
||||
}
|
||||
return s.operations[n].Move, nil
|
||||
}
|
||||
|
|
|
@ -324,6 +324,64 @@ func testForestTreeApply(t *testing.T, constructor func(t *testing.T) Forest) {
|
|||
})
|
||||
}
|
||||
|
||||
func TestForest_GetOpLog(t *testing.T) {
|
||||
for i := range providers {
|
||||
t.Run(providers[i].name, func(t *testing.T) {
|
||||
testForestTreeGetOpLog(t, providers[i].construct)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func testForestTreeGetOpLog(t *testing.T, constructor func(t *testing.T) Forest) {
|
||||
cid := cidtest.ID()
|
||||
treeID := "version"
|
||||
logs := []Move{
|
||||
{
|
||||
Meta: Meta{Time: 4, Items: []KeyValue{{"grand", []byte{1}}}},
|
||||
Child: 1,
|
||||
},
|
||||
{
|
||||
Meta: Meta{Time: 5, Items: []KeyValue{{"second", []byte{1, 2, 3}}}},
|
||||
Child: 4,
|
||||
},
|
||||
{
|
||||
Parent: 10,
|
||||
Meta: Meta{Time: 256 + 4, Items: []KeyValue{}}, // make sure keys are big-endian
|
||||
Child: 11,
|
||||
},
|
||||
}
|
||||
|
||||
s := constructor(t)
|
||||
|
||||
t.Run("empty log, no panic", func(t *testing.T) {
|
||||
_, err := s.TreeGetOpLog(cid, treeID, 0)
|
||||
require.ErrorIs(t, err, ErrTreeNotFound)
|
||||
})
|
||||
|
||||
for i := range logs {
|
||||
require.NoError(t, s.TreeApply(cid, treeID, &logs[i]))
|
||||
}
|
||||
|
||||
testGetOpLog := func(t *testing.T, height uint64, m Move) {
|
||||
lm, err := s.TreeGetOpLog(cid, treeID, height)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, m, lm)
|
||||
}
|
||||
|
||||
testGetOpLog(t, 0, logs[0])
|
||||
testGetOpLog(t, 4, logs[0])
|
||||
testGetOpLog(t, 5, logs[1])
|
||||
testGetOpLog(t, 6, logs[2])
|
||||
testGetOpLog(t, 260, logs[2])
|
||||
t.Run("missing entry", func(t *testing.T) {
|
||||
testGetOpLog(t, 261, Move{})
|
||||
})
|
||||
t.Run("missing tree", func(t *testing.T) {
|
||||
_, err := s.TreeGetOpLog(cid, treeID+"123", 4)
|
||||
require.ErrorIs(t, err, ErrTreeNotFound)
|
||||
})
|
||||
}
|
||||
|
||||
func TestForest_ApplyRandom(t *testing.T) {
|
||||
for i := range providers {
|
||||
t.Run(providers[i].name, func(t *testing.T) {
|
||||
|
|
|
@ -26,6 +26,9 @@ type Forest interface {
|
|||
// TreeGetChildren returns children of the node with the specified ID. The order is arbitrary.
|
||||
// Should return ErrTreeNotFound if the tree is not found, and empty result if the node is not in the tree.
|
||||
TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node) ([]uint64, error)
|
||||
// TreeGetOpLog returns first log operation stored at or above the height.
|
||||
// In case no such operation is found, empty Move and nil error should be returned.
|
||||
TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error)
|
||||
}
|
||||
|
||||
type ForestStorage interface {
|
||||
|
|
|
@ -36,3 +36,8 @@ func (s *Shard) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node)
|
|||
func (s *Shard) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.Node) ([]uint64, error) {
|
||||
return s.pilorama.TreeGetChildren(cid, treeID, nodeID)
|
||||
}
|
||||
|
||||
// TreeGetOpLog implements the pilorama.Forest interface.
|
||||
func (s *Shard) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
|
||||
return s.pilorama.TreeGetOpLog(cid, treeID, height)
|
||||
}
|
||||
|
|
|
@ -342,6 +342,38 @@ loop:
|
|||
})
|
||||
}
|
||||
|
||||
func (s *Service) GetOpLog(req *GetOpLogRequest, srv TreeService_GetOpLogServer) error {
|
||||
b := req.GetBody()
|
||||
|
||||
var cid cidSDK.ID
|
||||
if err := cid.Decode(req.GetBody().GetContainerId()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
h := b.GetHeight()
|
||||
for {
|
||||
lm, err := s.forest.TreeGetOpLog(cid, b.GetTreeId(), h)
|
||||
if err != nil || lm.Time == 0 {
|
||||
return err
|
||||
}
|
||||
|
||||
err = srv.Send(&GetOpLogResponse{
|
||||
Body: &GetOpLogResponse_Body{
|
||||
Operation: &LogMove{
|
||||
ParentId: lm.Parent,
|
||||
Meta: lm.Meta.Bytes(),
|
||||
ChildId: lm.Child,
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
h = lm.Time + 1
|
||||
}
|
||||
}
|
||||
|
||||
func protoToMeta(arr []*KeyValue) []pilorama.KeyValue {
|
||||
meta := make([]pilorama.KeyValue, len(arr))
|
||||
for i, kv := range arr {
|
||||
|
|
BIN
pkg/services/tree/service.pb.go
generated
BIN
pkg/services/tree/service.pb.go
generated
Binary file not shown.
|
@ -28,6 +28,7 @@ service TreeService {
|
|||
// Apply pushes log operation from another node to the current.
|
||||
// The request must be signed by a container node.
|
||||
rpc Apply (ApplyRequest) returns (ApplyResponse);
|
||||
rpc GetOpLog(GetOpLogRequest) returns (stream GetOpLogResponse);
|
||||
}
|
||||
|
||||
message AddRequest {
|
||||
|
@ -200,3 +201,25 @@ message ApplyResponse {
|
|||
Body body = 1;
|
||||
Signature signature = 2;
|
||||
};
|
||||
|
||||
|
||||
message GetOpLogRequest {
|
||||
message Body {
|
||||
bytes container_id = 1;
|
||||
string tree_id = 2;
|
||||
uint64 height = 3;
|
||||
uint64 count = 4;
|
||||
}
|
||||
|
||||
Body body = 1;
|
||||
Signature signature = 2;
|
||||
}
|
||||
|
||||
message GetOpLogResponse {
|
||||
message Body {
|
||||
LogMove operation = 1;
|
||||
}
|
||||
|
||||
Body body = 1;
|
||||
Signature signature = 2;
|
||||
};
|
||||
|
|
BIN
pkg/services/tree/service_grpc.pb.go
generated
BIN
pkg/services/tree/service_grpc.pb.go
generated
Binary file not shown.
BIN
pkg/services/tree/service_neofs.pb.go
generated
BIN
pkg/services/tree/service_neofs.pb.go
generated
Binary file not shown.
Loading…
Reference in a new issue