107 lines
2.8 KiB
Go
107 lines
2.8 KiB
Go
|
package sharding
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"testing"
|
||
|
|
|
||
|
|
"github.com/ipfs-cluster/ipfs-cluster/api"
|
||
|
|
)
|
||
|
|
|
||
|
|
// MockPinStore is used in VerifyShards
|
||
|
|
type MockPinStore interface {
|
||
|
|
// Gets a pin
|
||
|
|
PinGet(context.Context, api.Cid) (api.Pin, error)
|
||
|
|
}
|
||
|
|
|
||
|
|
// MockBlockStore is used in VerifyShards
|
||
|
|
type MockBlockStore interface {
|
||
|
|
// Gets a block
|
||
|
|
BlockGet(context.Context, api.Cid) ([]byte, error)
|
||
|
|
}
|
||
|
|
|
||
|
|
// VerifyShards checks that a sharded CID has been correctly formed and stored.
|
||
|
|
// This is a helper function for testing. It returns a map with all the blocks
|
||
|
|
// from all shards.
|
||
|
|
func VerifyShards(t *testing.T, rootCid api.Cid, pins MockPinStore, ipfs MockBlockStore, expectedShards int) (map[string]struct{}, error) {
|
||
|
|
ctx := context.Background()
|
||
|
|
metaPin, err := pins.PinGet(ctx, rootCid)
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("meta pin was not pinned: %s", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if api.PinType(metaPin.Type) != api.MetaType {
|
||
|
|
return nil, fmt.Errorf("bad MetaPin type")
|
||
|
|
}
|
||
|
|
|
||
|
|
if metaPin.Reference == nil {
|
||
|
|
return nil, errors.New("metaPin.Reference is unset")
|
||
|
|
}
|
||
|
|
|
||
|
|
clusterPin, err := pins.PinGet(ctx, *metaPin.Reference)
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("cluster pin was not pinned: %s", err)
|
||
|
|
}
|
||
|
|
if api.PinType(clusterPin.Type) != api.ClusterDAGType {
|
||
|
|
return nil, fmt.Errorf("bad ClusterDAGPin type")
|
||
|
|
}
|
||
|
|
|
||
|
|
if !clusterPin.Reference.Equals(metaPin.Cid) {
|
||
|
|
return nil, fmt.Errorf("clusterDAG should reference the MetaPin")
|
||
|
|
}
|
||
|
|
|
||
|
|
clusterDAGBlock, err := ipfs.BlockGet(ctx, clusterPin.Cid)
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("cluster pin was not stored: %s", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
clusterDAGNode, err := CborDataToNode(clusterDAGBlock, "cbor")
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
|
||
|
|
shards := clusterDAGNode.Links()
|
||
|
|
if len(shards) != expectedShards {
|
||
|
|
return nil, fmt.Errorf("bad number of shards")
|
||
|
|
}
|
||
|
|
|
||
|
|
shardBlocks := make(map[string]struct{})
|
||
|
|
var ref api.Cid
|
||
|
|
// traverse shards in order
|
||
|
|
for i := 0; i < len(shards); i++ {
|
||
|
|
sh, _, err := clusterDAGNode.ResolveLink([]string{fmt.Sprintf("%d", i)})
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
|
||
|
|
shardPin, err := pins.PinGet(ctx, api.NewCid(sh.Cid))
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("shard was not pinned: %s %s", sh.Cid, err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if ref != api.CidUndef && !shardPin.Reference.Equals(ref) {
|
||
|
|
t.Errorf("Ref (%s) should point to previous shard (%s)", ref, shardPin.Reference)
|
||
|
|
}
|
||
|
|
ref = shardPin.Cid
|
||
|
|
|
||
|
|
shardBlock, err := ipfs.BlockGet(ctx, shardPin.Cid)
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("shard block was not stored: %s", err)
|
||
|
|
}
|
||
|
|
shardNode, err := CborDataToNode(shardBlock, "cbor")
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
for _, l := range shardNode.Links() {
|
||
|
|
ci := l.Cid.String()
|
||
|
|
_, ok := shardBlocks[ci]
|
||
|
|
if ok {
|
||
|
|
return nil, fmt.Errorf("block belongs to two shards: %s", ci)
|
||
|
|
}
|
||
|
|
shardBlocks[ci] = struct{}{}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return shardBlocks, nil
|
||
|
|
}
|