mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-23 04:36:40 +00:00
fix: make DeleteVolume idempotent when a concurrent sweep already removed an object
deleteS3Prefix returned on any DeleteObject error, so two overlapping deletes for one volume left the loser reporting ServerInternal once the winner had removed a chunk between the list and the delete. DeleteDBInstance surfaced that as HTTP 500. S3ObjectStore.DeleteObject now reports a missing key as NoSuchKeyError, matching GetObject and HeadObject, and the sweep treats an already-deleted object as done. S3 answers 204 here but Predastore answers 404, and both mean the same thing to a delete sweep.
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
package handlers_ec2_volume
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/mulgadc/spinifex/spinifex/config"
|
||||
"github.com/mulgadc/spinifex/spinifex/objectstore"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func newPrefixTestConfig() *config.Config {
|
||||
return &config.Config{
|
||||
AZ: "ap-southeast-2a",
|
||||
Predastore: config.PredastoreConfig{
|
||||
Bucket: "test-bucket",
|
||||
Region: "ap-southeast-2",
|
||||
Host: fakeS3Host,
|
||||
AccessKey: "testkey",
|
||||
SecretKey: "testsecret",
|
||||
},
|
||||
WalDir: "/tmp/test-wal",
|
||||
}
|
||||
}
|
||||
|
||||
// racingDeleteStore answers every delete the way a backend does once a
|
||||
// concurrent sweep has already removed the key.
|
||||
type racingDeleteStore struct {
|
||||
objectstore.ObjectStore
|
||||
|
||||
deletes int
|
||||
err error
|
||||
}
|
||||
|
||||
func (r *racingDeleteStore) DeleteObject(_ context.Context, input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
|
||||
r.deletes++
|
||||
return nil, r.err
|
||||
}
|
||||
|
||||
func newPrefixTestStore(t *testing.T, keys ...string) *objectstore.MemoryObjectStore {
|
||||
t.Helper()
|
||||
store := objectstore.NewMemoryObjectStore()
|
||||
for _, key := range keys {
|
||||
_, err := store.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String("test-bucket"),
|
||||
Key: aws.String(key),
|
||||
Body: strings.NewReader("chunk"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
return store
|
||||
}
|
||||
|
||||
func TestDeleteS3PrefixTreatsMissingObjectAsDeleted(t *testing.T) {
|
||||
backing := newPrefixTestStore(t, "vol-abc/chunks/chunk.00000028.bin", "vol-abc/config.json")
|
||||
store := &racingDeleteStore{
|
||||
ObjectStore: backing,
|
||||
err: &objectstore.NoSuchKeyError{Key: "vol-abc/chunks/chunk.00000028.bin"},
|
||||
}
|
||||
svc := NewVolumeServiceImplWithStore(newPrefixTestConfig(), store, nil)
|
||||
|
||||
err := svc.deleteS3Prefix(context.Background(), "vol-abc/")
|
||||
|
||||
require.NoError(t, err, "an object another sweep already removed must not fail the delete")
|
||||
assert.Equal(t, 2, store.deletes, "the sweep must continue past a missing object")
|
||||
}
|
||||
|
||||
func TestDeleteS3PrefixPropagatesBackendFailure(t *testing.T) {
|
||||
backing := newPrefixTestStore(t, "vol-abc/chunks/chunk.00000028.bin")
|
||||
store := &racingDeleteStore{ObjectStore: backing, err: errors.New("connection refused")}
|
||||
svc := NewVolumeServiceImplWithStore(newPrefixTestConfig(), store, nil)
|
||||
|
||||
err := svc.deleteS3Prefix(context.Background(), "vol-abc/")
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "connection refused")
|
||||
}
|
||||
|
||||
func TestDeleteS3PrefixRemovesEveryObject(t *testing.T) {
|
||||
store := newPrefixTestStore(t, "vol-abc/chunks/chunk.00000000.bin", "vol-abc/config.json")
|
||||
svc := NewVolumeServiceImplWithStore(newPrefixTestConfig(), store, nil)
|
||||
|
||||
require.NoError(t, svc.deleteS3Prefix(context.Background(), "vol-abc/"))
|
||||
|
||||
out, err := store.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{
|
||||
Bucket: aws.String("test-bucket"),
|
||||
Prefix: aws.String("vol-abc/"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, out.Contents)
|
||||
}
|
||||
@@ -1609,7 +1609,9 @@ func (s *VolumeServiceImpl) deleteS3Prefix(ctx context.Context, prefix string) e
|
||||
Bucket: aws.String(bucket),
|
||||
Key: obj.Key,
|
||||
})
|
||||
if err != nil {
|
||||
// A concurrent sweep may have deleted the object between the list
|
||||
// and here. It is gone either way, which is what this asked for.
|
||||
if err != nil && !objectstore.IsNoSuchKeyError(err) {
|
||||
return fmt.Errorf("failed to delete object %s: %w", *obj.Key, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
package objectstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const noSuchKeyBody = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Error><Code>NoSuchKey</Code><Message>The specified key does not exist</Message></Error>`
|
||||
|
||||
// deleteResponder answers every DELETE with a fixed status and body, standing in
|
||||
// for a backend that has already dropped the key.
|
||||
type deleteResponder struct {
|
||||
code int
|
||||
body string
|
||||
}
|
||||
|
||||
func (d *deleteResponder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodDelete {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(d.code)
|
||||
if d.body != "" {
|
||||
_, _ = w.Write([]byte(d.body))
|
||||
}
|
||||
}
|
||||
|
||||
func TestS3ObjectStoreDeleteObjectMissingKey(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
code int
|
||||
body string
|
||||
}{
|
||||
{name: "NoSuchKey body", code: http.StatusNotFound, body: noSuchKeyBody},
|
||||
{name: "bare 404", code: http.StatusNotFound},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
srv := httptest.NewServer(&deleteResponder{code: tt.code, body: tt.body})
|
||||
defer srv.Close()
|
||||
|
||||
_, err := newS3Store(t, srv.URL).DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String("bucket"),
|
||||
Key: aws.String("vol-abc/chunks/chunk.00000028.bin"),
|
||||
})
|
||||
|
||||
require.Error(t, err)
|
||||
assert.True(t, IsNoSuchKeyError(err), "want NoSuchKeyError, got %v", err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestS3ObjectStoreDeleteObjectPropagatesOtherErrors(t *testing.T) {
|
||||
srv := httptest.NewServer(&deleteResponder{code: http.StatusInternalServerError})
|
||||
defer srv.Close()
|
||||
|
||||
_, err := newS3Store(t, srv.URL).DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String("bucket"),
|
||||
Key: aws.String("vol-abc/chunks/chunk.00000028.bin"),
|
||||
})
|
||||
|
||||
require.Error(t, err)
|
||||
assert.False(t, IsNoSuchKeyError(err), "a backend failure must not read as a missing key")
|
||||
}
|
||||
|
||||
func TestS3ObjectStoreDeleteObjectSucceeds(t *testing.T) {
|
||||
srv := httptest.NewServer(&deleteResponder{code: http.StatusNoContent})
|
||||
defer srv.Close()
|
||||
|
||||
_, err := newS3Store(t, srv.URL).DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String("bucket"),
|
||||
Key: aws.String("vol-abc/chunks/chunk.00000028.bin"),
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
}
|
||||
@@ -119,8 +119,20 @@ func (s *S3ObjectStore) PutObject(ctx context.Context, input *s3.PutObjectInput)
|
||||
return s.client.PutObjectWithContext(ctx, input)
|
||||
}
|
||||
|
||||
// DeleteObject reports a missing key as NoSuchKeyError so callers can tell an
|
||||
// already-deleted object from a backend failure. S3 itself answers 204 here, but
|
||||
// Predastore answers 404, and a delete sweep must treat both as done.
|
||||
func (s *S3ObjectStore) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
|
||||
return s.client.DeleteObjectWithContext(ctx, input)
|
||||
out, err := s.client.DeleteObjectWithContext(ctx, input)
|
||||
if err != nil {
|
||||
var aerr awserr.Error
|
||||
if errors.As(err, &aerr) &&
|
||||
(aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound") {
|
||||
return nil, &NoSuchKeyError{Key: aws.StringValue(input.Key)}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *S3ObjectStore) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
|
||||
|
||||
Reference in New Issue
Block a user