022f5d2e14
* add cache, update cache w/piece create/delete * add service w/loop to cache to recalculate space used cache * add piecestore cache to other sn svcs to use * add table to persist the total space used * rm cache where not needed * rm stuff from sn svcs * start fixing tests, changes per comments * update commits * add unit tests * fix commiting before we write header bytes * fix cache create test * copy cache map, add started back to recalc * fix test * add test, update comments
438 lines
15 KiB
Go
438 lines
15 KiB
Go
// Copyright (C) 2019 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package pieces_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"io"
|
|
"io/ioutil"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"go.uber.org/zap/zaptest"
|
|
|
|
"storj.io/storj/internal/memory"
|
|
"storj.io/storj/internal/testcontext"
|
|
"storj.io/storj/internal/testidentity"
|
|
"storj.io/storj/internal/testrand"
|
|
"storj.io/storj/pkg/pb"
|
|
"storj.io/storj/pkg/pkcrypto"
|
|
"storj.io/storj/pkg/storj"
|
|
"storj.io/storj/storage"
|
|
"storj.io/storj/storage/filestore"
|
|
"storj.io/storj/storagenode"
|
|
"storj.io/storj/storagenode/pieces"
|
|
"storj.io/storj/storagenode/storagenodedb/storagenodedbtest"
|
|
)
|
|
|
|
func TestPieces(t *testing.T) {
|
|
ctx := testcontext.New(t)
|
|
defer ctx.Cleanup()
|
|
|
|
dir, err := filestore.NewDir(ctx.Dir("pieces"))
|
|
require.NoError(t, err)
|
|
|
|
blobs := filestore.New(zaptest.NewLogger(t), dir)
|
|
defer ctx.Check(blobs.Close)
|
|
|
|
store := pieces.NewStore(zaptest.NewLogger(t), blobs, nil, nil, nil)
|
|
|
|
satelliteID := testidentity.MustPregeneratedSignedIdentity(0, storj.LatestIDVersion()).ID
|
|
pieceID := storj.NewPieceID()
|
|
|
|
source := testrand.Bytes(8000)
|
|
|
|
{ // write data
|
|
writer, err := store.Writer(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
|
|
n, err := io.Copy(writer, bytes.NewReader(source))
|
|
require.NoError(t, err)
|
|
assert.Equal(t, len(source), int(n))
|
|
assert.Equal(t, len(source), int(writer.Size()))
|
|
|
|
// verify hash
|
|
hash := pkcrypto.NewHash()
|
|
_, _ = hash.Write(source)
|
|
assert.Equal(t, hash.Sum(nil), writer.Hash())
|
|
|
|
// commit
|
|
require.NoError(t, writer.Commit(ctx, &pb.PieceHeader{}))
|
|
// after commit we should be able to call cancel without an error
|
|
require.NoError(t, writer.Cancel(ctx))
|
|
}
|
|
|
|
{ // valid reads
|
|
read := func(offset, length int64) []byte {
|
|
reader, err := store.Reader(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
|
|
pos, err := reader.Seek(offset, io.SeekStart)
|
|
require.NoError(t, err)
|
|
require.Equal(t, offset, pos)
|
|
|
|
data := make([]byte, length)
|
|
n, err := io.ReadFull(reader, data)
|
|
require.NoError(t, err)
|
|
require.Equal(t, int(length), n)
|
|
|
|
require.NoError(t, reader.Close())
|
|
|
|
return data
|
|
}
|
|
|
|
require.Equal(t, source[10:11], read(10, 1))
|
|
require.Equal(t, source[10:1010], read(10, 1000))
|
|
require.Equal(t, source, read(0, int64(len(source))))
|
|
}
|
|
|
|
{ // reading ends with io.EOF
|
|
reader, err := store.Reader(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
|
|
data := make([]byte, 111)
|
|
for {
|
|
_, err := reader.Read(data)
|
|
if err != nil {
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
require.NoError(t, err)
|
|
}
|
|
}
|
|
|
|
require.NoError(t, reader.Close())
|
|
}
|
|
|
|
{ // test delete
|
|
assert.NoError(t, store.Delete(ctx, satelliteID, pieceID))
|
|
// read should now fail
|
|
_, err := store.Reader(ctx, satelliteID, pieceID)
|
|
assert.Error(t, err)
|
|
}
|
|
|
|
{ // write cancel
|
|
cancelledPieceID := storj.NewPieceID()
|
|
writer, err := store.Writer(ctx, satelliteID, cancelledPieceID)
|
|
require.NoError(t, err)
|
|
|
|
n, err := io.Copy(writer, bytes.NewReader(source))
|
|
require.NoError(t, err)
|
|
assert.Equal(t, len(source), int(n))
|
|
assert.Equal(t, len(source), int(writer.Size()))
|
|
|
|
// cancel writing
|
|
require.NoError(t, writer.Cancel(ctx))
|
|
// commit should not fail
|
|
require.Error(t, writer.Commit(ctx, &pb.PieceHeader{}))
|
|
|
|
// read should fail
|
|
_, err = store.Reader(ctx, satelliteID, cancelledPieceID)
|
|
assert.Error(t, err)
|
|
}
|
|
}
|
|
|
|
func writeAPiece(ctx context.Context, t testing.TB, store *pieces.Store, satelliteID storj.NodeID, pieceID storj.PieceID, data []byte, atTime time.Time, expireTime *time.Time, formatVersion storage.FormatVersion) {
|
|
tStore := &pieces.StoreForTest{store}
|
|
writer, err := tStore.WriterForFormatVersion(ctx, satelliteID, pieceID, formatVersion)
|
|
require.NoError(t, err)
|
|
|
|
_, err = writer.Write(data)
|
|
require.NoError(t, err)
|
|
size := writer.Size()
|
|
assert.Equal(t, int64(len(data)), size)
|
|
limit := pb.OrderLimit{}
|
|
if expireTime != nil {
|
|
limit.PieceExpiration = *expireTime
|
|
}
|
|
err = writer.Commit(ctx, &pb.PieceHeader{
|
|
Hash: writer.Hash(),
|
|
CreationTime: atTime,
|
|
OrderLimit: limit,
|
|
})
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func verifyPieceHandle(t testing.TB, reader *pieces.Reader, expectDataLen int, expectCreateTime time.Time, expectFormat storage.FormatVersion) {
|
|
assert.Equal(t, expectFormat, reader.StorageFormatVersion())
|
|
assert.Equal(t, int64(expectDataLen), reader.Size())
|
|
if expectFormat != filestore.FormatV0 {
|
|
pieceHeader, err := reader.GetPieceHeader()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, expectFormat, storage.FormatVersion(pieceHeader.FormatVersion))
|
|
assert.Equal(t, expectCreateTime.UTC(), pieceHeader.CreationTime.UTC())
|
|
}
|
|
}
|
|
|
|
func tryOpeningAPiece(ctx context.Context, t testing.TB, store *pieces.Store, satelliteID storj.NodeID, pieceID storj.PieceID, expectDataLen int, expectTime time.Time, expectFormat storage.FormatVersion) {
|
|
reader, err := store.Reader(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
verifyPieceHandle(t, reader, expectDataLen, expectTime, expectFormat)
|
|
require.NoError(t, reader.Close())
|
|
|
|
reader, err = store.ReaderWithStorageFormat(ctx, satelliteID, pieceID, expectFormat)
|
|
require.NoError(t, err)
|
|
verifyPieceHandle(t, reader, expectDataLen, expectTime, expectFormat)
|
|
require.NoError(t, reader.Close())
|
|
}
|
|
|
|
// Test that the piece store can still read V0 pieces that might be left over from a previous
|
|
// version, as well as V1 pieces.
|
|
func TestMultipleStorageFormatVersions(t *testing.T) {
|
|
ctx := testcontext.New(t)
|
|
defer ctx.Cleanup()
|
|
|
|
blobs, err := filestore.NewAt(zaptest.NewLogger(t), ctx.Dir("store"))
|
|
require.NoError(t, err)
|
|
defer ctx.Check(blobs.Close)
|
|
|
|
store := pieces.NewStore(zaptest.NewLogger(t), blobs, nil, nil, nil)
|
|
|
|
const pieceSize = 1024
|
|
|
|
var (
|
|
data = testrand.Bytes(pieceSize)
|
|
satellite = testrand.NodeID()
|
|
v0PieceID = testrand.PieceID()
|
|
v1PieceID = testrand.PieceID()
|
|
now = time.Now().UTC()
|
|
)
|
|
|
|
// write a V0 piece
|
|
writeAPiece(ctx, t, store, satellite, v0PieceID, data, now, nil, filestore.FormatV0)
|
|
|
|
// write a V1 piece
|
|
writeAPiece(ctx, t, store, satellite, v1PieceID, data, now, nil, filestore.FormatV1)
|
|
|
|
// look up the different pieces with Reader and ReaderWithStorageFormat
|
|
tryOpeningAPiece(ctx, t, store, satellite, v0PieceID, len(data), now, filestore.FormatV0)
|
|
tryOpeningAPiece(ctx, t, store, satellite, v1PieceID, len(data), now, filestore.FormatV1)
|
|
|
|
// write a V1 piece with the same ID as the V0 piece (to simulate it being rewritten as
|
|
// V1 during a migration)
|
|
differentData := append(data, 111, 104, 97, 105)
|
|
writeAPiece(ctx, t, store, satellite, v0PieceID, differentData, now, nil, filestore.FormatV1)
|
|
|
|
// if we try to access the piece at that key, we should see only the V1 piece
|
|
tryOpeningAPiece(ctx, t, store, satellite, v0PieceID, len(differentData), now, filestore.FormatV1)
|
|
|
|
// unless we ask specifically for a V0 piece
|
|
reader, err := store.ReaderWithStorageFormat(ctx, satellite, v0PieceID, filestore.FormatV0)
|
|
require.NoError(t, err)
|
|
verifyPieceHandle(t, reader, len(data), now, filestore.FormatV0)
|
|
require.NoError(t, reader.Close())
|
|
|
|
// delete the v0PieceID; both the V0 and the V1 pieces should go away
|
|
err = store.Delete(ctx, satellite, v0PieceID)
|
|
require.NoError(t, err)
|
|
|
|
reader, err = store.Reader(ctx, satellite, v0PieceID)
|
|
require.Error(t, err)
|
|
require.True(t, os.IsNotExist(err))
|
|
assert.Nil(t, reader)
|
|
}
|
|
|
|
func TestGetExpired(t *testing.T) {
|
|
storagenodedbtest.Run(t, func(t *testing.T, db storagenode.DB) {
|
|
ctx := testcontext.New(t)
|
|
defer ctx.Cleanup()
|
|
|
|
v0PieceInfo, ok := db.V0PieceInfo().(pieces.V0PieceInfoDBForTest)
|
|
require.True(t, ok, "V0PieceInfoDB can not satisfy V0PieceInfoDBForTest")
|
|
expirationInfo := db.PieceExpirationDB()
|
|
|
|
store := pieces.NewStore(zaptest.NewLogger(t), db.Pieces(), v0PieceInfo, expirationInfo, db.PieceSpaceUsedDB())
|
|
|
|
now := time.Now().UTC()
|
|
testDates := []struct {
|
|
years, months, days int
|
|
}{
|
|
{-20, -1, -2},
|
|
{1, 6, 14},
|
|
{0, -1, 0},
|
|
{0, 0, 1},
|
|
}
|
|
testPieces := make([]pieces.Info, 4)
|
|
for p := range testPieces {
|
|
testPieces[p] = pieces.Info{
|
|
SatelliteID: testrand.NodeID(),
|
|
PieceID: testrand.PieceID(),
|
|
OrderLimit: &pb.OrderLimit{},
|
|
UplinkPieceHash: &pb.PieceHash{},
|
|
PieceExpiration: now.AddDate(testDates[p].years, testDates[p].months, testDates[p].days),
|
|
}
|
|
}
|
|
|
|
// put testPieces 0 and 1 in the v0 pieceinfo db
|
|
err := v0PieceInfo.Add(ctx, &testPieces[0])
|
|
require.NoError(t, err)
|
|
err = v0PieceInfo.Add(ctx, &testPieces[1])
|
|
require.NoError(t, err)
|
|
|
|
// put testPieces 2 and 3 in the piece_expirations db
|
|
err = expirationInfo.SetExpiration(ctx, testPieces[2].SatelliteID, testPieces[2].PieceID, testPieces[2].PieceExpiration)
|
|
require.NoError(t, err)
|
|
err = expirationInfo.SetExpiration(ctx, testPieces[3].SatelliteID, testPieces[3].PieceID, testPieces[3].PieceExpiration)
|
|
require.NoError(t, err)
|
|
|
|
// GetExpired with limit 0 gives empty result
|
|
expired, err := store.GetExpired(ctx, now, 0)
|
|
require.NoError(t, err)
|
|
assert.Empty(t, expired)
|
|
|
|
// GetExpired with limit 1 gives only 1 result, although there are 2 possible
|
|
expired, err = store.GetExpired(ctx, now, 1)
|
|
require.NoError(t, err)
|
|
require.Len(t, expired, 1)
|
|
assert.Equal(t, testPieces[2].PieceID, expired[0].PieceID)
|
|
assert.Equal(t, testPieces[2].SatelliteID, expired[0].SatelliteID)
|
|
assert.False(t, expired[0].InPieceInfo)
|
|
|
|
// GetExpired with 2 or more gives all expired results correctly; one from
|
|
// piece_expirations, and one from pieceinfo
|
|
expired, err = store.GetExpired(ctx, now, 1000)
|
|
require.NoError(t, err)
|
|
require.Len(t, expired, 2)
|
|
assert.Equal(t, testPieces[2].PieceID, expired[0].PieceID)
|
|
assert.Equal(t, testPieces[2].SatelliteID, expired[0].SatelliteID)
|
|
assert.False(t, expired[0].InPieceInfo)
|
|
assert.Equal(t, testPieces[0].PieceID, expired[1].PieceID)
|
|
assert.Equal(t, testPieces[0].SatelliteID, expired[1].SatelliteID)
|
|
assert.True(t, expired[1].InPieceInfo)
|
|
})
|
|
}
|
|
|
|
func TestOverwriteV0WithV1(t *testing.T) {
|
|
storagenodedbtest.Run(t, func(t *testing.T, db storagenode.DB) {
|
|
ctx := testcontext.New(t)
|
|
defer ctx.Cleanup()
|
|
|
|
v0PieceInfo, ok := db.V0PieceInfo().(pieces.V0PieceInfoDBForTest)
|
|
require.True(t, ok, "V0PieceInfoDB can not satisfy V0PieceInfoDBForTest")
|
|
expirationInfo := db.PieceExpirationDB()
|
|
|
|
store := pieces.NewStore(zaptest.NewLogger(t), db.Pieces(), v0PieceInfo, expirationInfo, db.PieceSpaceUsedDB())
|
|
|
|
satelliteID := testrand.NodeID()
|
|
pieceID := testrand.PieceID()
|
|
v0Data := testrand.Bytes(4 * memory.MiB)
|
|
v1Data := testrand.Bytes(3 * memory.MiB)
|
|
|
|
// write the piece as V0. We can't provide the expireTime via writeAPiece, because
|
|
// BlobWriter.Commit only knows how to store expiration times in piece_expirations.
|
|
v0CreateTime := time.Now().UTC()
|
|
v0ExpireTime := v0CreateTime.AddDate(5, 0, 0)
|
|
writeAPiece(ctx, t, store, satelliteID, pieceID, v0Data, v0CreateTime, nil, filestore.FormatV0)
|
|
// now put the piece in the pieceinfo db directly, because store won't do that for us.
|
|
// this is where the expireTime takes effect.
|
|
err := v0PieceInfo.Add(ctx, &pieces.Info{
|
|
SatelliteID: satelliteID,
|
|
PieceID: pieceID,
|
|
PieceSize: int64(len(v0Data)),
|
|
PieceCreation: v0CreateTime,
|
|
PieceExpiration: v0ExpireTime,
|
|
OrderLimit: &pb.OrderLimit{},
|
|
UplinkPieceHash: &pb.PieceHash{},
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// ensure we can see it via store.Reader
|
|
{
|
|
reader, err := store.Reader(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(len(v0Data)), reader.Size())
|
|
assert.Equal(t, filestore.FormatV0, reader.StorageFormatVersion())
|
|
gotData, err := ioutil.ReadAll(reader)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, v0Data, gotData)
|
|
require.NoError(t, reader.Close())
|
|
}
|
|
|
|
// ensure we can see it via WalkSatellitePieces
|
|
calledTimes := 0
|
|
err = store.WalkSatellitePieces(ctx, satelliteID, func(access pieces.StoredPieceAccess) error {
|
|
calledTimes++
|
|
require.Equal(t, 1, calledTimes)
|
|
gotCreateTime, err := access.CreationTime(ctx)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, v0CreateTime, gotCreateTime)
|
|
gotSize, err := access.ContentSize(ctx)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(len(v0Data)), gotSize)
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// now "overwrite" the piece (write a new blob with the same id, but with V1 storage)
|
|
v1CreateTime := time.Now().UTC()
|
|
v1ExpireTime := v1CreateTime.AddDate(5, 0, 0)
|
|
writeAPiece(ctx, t, store, satelliteID, pieceID, v1Data, v1CreateTime, &v1ExpireTime, filestore.FormatV1)
|
|
|
|
// ensure we can see it (the new piece) via store.Reader
|
|
{
|
|
reader, err := store.Reader(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(len(v1Data)), reader.Size())
|
|
assert.Equal(t, filestore.FormatV1, reader.StorageFormatVersion())
|
|
gotData, err := ioutil.ReadAll(reader)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, v1Data, gotData)
|
|
require.NoError(t, reader.Close())
|
|
}
|
|
|
|
// now _both_ pieces should show up under WalkSatellitePieces. this may
|
|
// be counter-intuitive, but the V0 piece still exists for now (so we can avoid
|
|
// hitting the pieceinfo db with every new piece write). I believe this is OK, because
|
|
// (a) I don't think that writing different pieces with the same piece ID is a normal
|
|
// use case, unless we make a V0->V1 migrator tool, which should know about these
|
|
// semantics; (b) the V0 piece should not ever become visible again to the user; it
|
|
// should not be possible under normal conditions to delete one without deleting the
|
|
// other.
|
|
calledTimes = 0
|
|
err = store.WalkSatellitePieces(ctx, satelliteID, func(access pieces.StoredPieceAccess) error {
|
|
calledTimes++
|
|
switch calledTimes {
|
|
case 1:
|
|
// expect the V1 piece
|
|
assert.Equal(t, pieceID, access.PieceID())
|
|
assert.Equal(t, filestore.FormatV1, access.StorageFormatVersion())
|
|
gotCreateTime, err := access.CreationTime(ctx)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, v1CreateTime, gotCreateTime)
|
|
gotSize, err := access.ContentSize(ctx)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(len(v1Data)), gotSize)
|
|
case 2:
|
|
// expect the V0 piece
|
|
assert.Equal(t, pieceID, access.PieceID())
|
|
assert.Equal(t, filestore.FormatV0, access.StorageFormatVersion())
|
|
gotCreateTime, err := access.CreationTime(ctx)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, v0CreateTime, gotCreateTime)
|
|
gotSize, err := access.ContentSize(ctx)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(len(v0Data)), gotSize)
|
|
default:
|
|
t.Fatalf("calledTimes should be 1 or 2, but it is %d", calledTimes)
|
|
}
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// delete the pieceID; this should get both V0 and V1
|
|
err = store.Delete(ctx, satelliteID, pieceID)
|
|
require.NoError(t, err)
|
|
|
|
err = store.WalkSatellitePieces(ctx, satelliteID, func(access pieces.StoredPieceAccess) error {
|
|
t.Fatalf("this should not have been called. pieceID=%x, format=%d", access.PieceID(), access.StorageFormatVersion())
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
})
|
|
}
|