JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

uplink: don't stop deleting segments on first error (#2943)

Browse Source
This commit is contained in:
Michal Niewrzal 2019-09-05 14:25:30 +02:00 committed by GitHub
parent d6bbc2a653
commit 61168493dc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 42 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
internal/testplanet/uplink_test.go
View File

@ -26,6 +26,7 @@ import (
"storj.io/storj/pkg/server"
"storj.io/storj/pkg/storj"
"storj.io/storj/uplink"
"storj.io/storj/uplink/metainfo"
)
func TestUplinksParallel(t *testing.T) {
@ -245,3 +246,36 @@ func TestDownloadFromUnresponsiveNode(t *testing.T) {
assert.Equal(t, expectedData, data)
})
}
func TestDeleteWithOfflineStoragenode(t *testing.T) {
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1, StorageNodeCount: 6, UplinkCount: 1,
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
expectedData := testrand.Bytes(5 * memory.MiB)
config := planet.Uplinks[0].GetConfig(planet.Satellites[0])
config.Client.SegmentSize = 1 * memory.MiB
err := planet.Uplinks[0].UploadWithClientConfig(ctx, planet.Satellites[0], config, "test-bucket", "test-file", expectedData)
require.NoError(t, err)
for _, node := range planet.StorageNodes {
err = planet.StopPeer(node)
require.NoError(t, err)
}
err = planet.Uplinks[0].Delete(ctx, planet.Satellites[0], "test-bucket", "test-file")
require.Error(t, err)
apiKey := planet.Uplinks[0].APIKey[planet.Satellites[0].ID()]
metainfoClient, err := planet.Uplinks[0].DialMetainfo(ctx, planet.Satellites[0], apiKey)
require.NoError(t, err)
defer ctx.Check(metainfoClient.Close)
objects, _, err := metainfoClient.ListObjects(ctx, metainfo.ListObjectsParams{
Bucket: []byte("test-bucket"),
})
require.NoError(t, err)
require.Equal(t, 0, len(objects))
})
}

11
uplink/storage/streams/store.go
View File

@ -437,19 +437,24 @@ func (s *streamStore) Delete(ctx context.Context, path Path, pathCipher storj.Ci
return err
}
var errlist errs.Group
for i := 0; i < int(numberOfSegments(&stream, &streamMeta)-1); i++ {
currentPath, err := createSegmentPath(ctx, int64(i), path.Bucket(), encPath)
if err != nil {
return err
errlist.Add(err)
continue
}
err = s.segments.Delete(ctx, currentPath)
if err != nil {
return err
errlist.Add(err)
continue
}
}
return s.segments.Delete(ctx, lastSegmentPath)
errlist.Add(s.segments.Delete(ctx, lastSegmentPath))
return errlist.Err()
}
// ListItem is a single item in a listing