mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-17 20:57:27 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
seaweedfs/weed/s3api/s3tables/handler_rename_test.go
T
Chris LuandGitHub d044839ab2 iceberg: make a table commit a compare-and-swap (#10775) 
* iceberg: make a table commit a compare-and-swap

The catalog validated the caller's version token, ran its authorization
checks, and only then wrote the new metadata xattr. Two engines
committing against the same base both passed that check and both wrote,
so the second silently dropped the first one's snapshot. Both also derive
the same v{N}.metadata.json name and the file write overwrote, leaving
the surviving pointer aimed at the loser's metadata - and the loser's
conflict cleanup then deleted the winner's file.

Write the metadata file with an exclusive create and update the xattr
conditionally on the bytes the handler read, the way the maintenance
worker already commits. A writer that lost the race re-reads and retries,
and reports 409 CommitFailedException once out of attempts.

* iceberg: stage a commit under a unique name when the versioned one is taken

Two follow-ups from review of the commit compare-and-swap:

Refusing to overwrite v{N}.metadata.json also refused to get past a file
left behind by a commit that died between staging and updating the
pointer. Every later commit derived the same name, saw the collision, and
reported a conflict, so the table stayed uncommittable until an orphan
sweep removed the file. Stage under v{N}-{uuid} instead: neither writer's
file is overwritten and the catalog pointer still decides who won, which
is how the maintenance worker has always staged its own metadata.
metadataVersionFromLocation learned to read the version back out of that
name.

The conditional update guarded only the metadata attribute while the
write replaced the whole entry, so a policy or tag written in the same
window was silently reverted. Guard every catalog attribute, which turns
that into a conflict the caller retries on fresh state.

* iceberg: give saveMetadataFile the exclusive flag instead of a second name

saveNewMetadataFile, saveMetadataBlobExclusive and uniqueMetadataFileName
were three new names around one existing helper. The flag now rides on
saveMetadataFile and saveMetadataBlob, and the unique-name construction
sits where it is used.

* iceberg: reuse the filer CAS helpers #10773 added, and stage transactions exclusively

#10773 landed mutateEntryExtended, which already writes an entry back under a
whole-entry precondition and retries. Drop the helper this branch added and
route the table commit through it: the check that the metadata is still the
one this request read now lives in the mutation, where it sees current state.

The policy the request was authorized against is asserted too, so an
administrator restricting it mid-commit sends the caller back through
authorization instead of having a stale decision applied. Bucket and
namespace policies live on other entries and a single-entry precondition
cannot cover them.

Multi-table transactions stage their metadata exclusively for the same
reason single-table commits do, and carry the name they landed on into the
pointer flip.
2026-08-16 12:55:42 -07:00

437 lines
16 KiB
Go
Raw Permalink Blame History

package s3tables
import (
"bytes"
"context"
"encoding/json"
"net"
"path"
"sort"
"strings"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
)
// memFilerServer is an in-memory filer used to drive Manager operations
// end-to-end without a live cluster.
type memFilerServer struct {
filer_pb.UnimplementedSeaweedFilerServer
entries map[string]map[string]*filer_pb.Entry // dir -> name -> entry
client filer_pb.SeaweedFilerClient
// beforeUpdate runs once, at the start of the next UpdateEntry, so a test
// can land a competing write in a handler's read-to-write window.
beforeUpdate func()
}
func newMemFilerServer() *memFilerServer {
return &memFilerServer{entries: make(map[string]map[string]*filer_pb.Entry)}
}
func (f *memFilerServer) getEntry(dir, name string) *filer_pb.Entry {
if d, ok := f.entries[dir]; ok {
return d[name]
}
return nil
}
func (f *memFilerServer) putEntry(dir, name string, extended map[string][]byte) {
if _, ok := f.entries[dir]; !ok {
f.entries[dir] = make(map[string]*filer_pb.Entry)
}
f.entries[dir][name] = &filer_pb.Entry{Name: name, IsDirectory: true, Extended: extended}
}
func (f *memFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
if e := f.getEntry(req.Directory, req.Name); e != nil {
return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil
}
// Carry the sentinel text so filer_pb.LookupEntry maps it to ErrNotFound.
return nil, status.Errorf(codes.NotFound, "%s: %s/%s", filer_pb.ErrNotFound.Error(), req.Directory, req.Name)
}
func (f *memFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error {
d, ok := f.entries[req.Directory]
if !ok {
return nil
}
names := make([]string, 0, len(d))
for name := range d {
names = append(names, name)
}
sort.Strings(names)
for _, name := range names {
if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: d[name]}); err != nil {
return err
}
}
return nil
}
func (f *memFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
if _, ok := f.entries[req.Directory]; !ok {
f.entries[req.Directory] = make(map[string]*filer_pb.Entry)
}
f.entries[req.Directory][req.Entry.Name] = req.Entry
return &filer_pb.CreateEntryResponse{}, nil
}
func (f *memFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
if hook := f.beforeUpdate; hook != nil {
f.beforeUpdate = nil
hook()
}
// The real filer validates ExpectedExtended under the per-path lock; without
// it here a lost update would look like a success.
for key, expected := range req.ExpectedExtended {
var actual []byte
if existing := f.getEntry(req.Directory, req.Entry.Name); existing != nil {
actual = existing.Extended[key]
}
if !bytes.Equal(actual, expected) {
return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key)
}
}
if _, ok := f.entries[req.Directory]; !ok {
f.entries[req.Directory] = make(map[string]*filer_pb.Entry)
}
f.entries[req.Directory][req.Entry.Name] = req.Entry
return &filer_pb.UpdateEntryResponse{}, nil
}
func (f *memFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) {
if d, ok := f.entries[req.Directory]; ok {
delete(d, req.Name)
}
// Honor recursive data deletion so a regression that wipes the table directory
// also drops its metadata/ and data/ children (the data-loss this guards against).
if req.IsRecursive && req.IsDeleteData {
child := path.Join(req.Directory, req.Name)
for dir := range f.entries {
if dir == child || strings.HasPrefix(dir, child+"/") {
delete(f.entries, dir)
}
}
}
return &filer_pb.DeleteEntryResponse{}, nil
}
func (f *memFilerServer) Ping(_ context.Context, _ *filer_pb.PingRequest) (*filer_pb.PingResponse, error) {
now := time.Now().UnixNano()
return &filer_pb.PingResponse{StartTimeNs: now, RemoteTimeNs: now, StopTimeNs: now}, nil
}
func startMemFiler(t *testing.T) *memFilerServer {
t.Helper()
fs := newMemFilerServer()
listener, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
server := grpc.NewServer()
filer_pb.RegisterSeaweedFilerServer(server, fs)
go func() { _ = server.Serve(listener) }()
t.Cleanup(server.GracefulStop)
conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
require.NoError(t, err)
t.Cleanup(func() { _ = conn.Close() })
fs.client = filer_pb.NewSeaweedFilerClient(conn)
deadline := time.Now().Add(5 * time.Second)
for {
pingCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
_, err := fs.client.Ping(pingCtx, &filer_pb.PingRequest{})
cancel()
if err == nil {
break
}
require.False(t, time.Now().After(deadline), "filer not ready: %v", err)
time.Sleep(10 * time.Millisecond)
}
return fs
}
const renameTestBucket = "renamebkt"
func mustBucketARN(t *testing.T) string {
t.Helper()
arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, renameTestBucket)
require.NoError(t, err)
return arn
}
// startRenameManager seeds a bucket/namespace/table and returns a trusted Manager.
func startRenameManager(t *testing.T) (*memFilerServer, *Manager) {
t.Helper()
fs := startMemFiler(t)
bucketMeta, _ := json.Marshal(tableBucketMetadata{Name: renameTestBucket, OwnerAccountID: DefaultAccountID})
fs.putEntry(TablesPath, renameTestBucket, map[string][]byte{
ExtendedKeyTableBucket: []byte("{}"),
ExtendedKeyMetadata: bucketMeta,
})
nsMeta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"ns"}, OwnerAccountID: DefaultAccountID})
fs.putEntry(GetTableBucketPath(renameTestBucket), "ns", map[string][]byte{ExtendedKeyMetadata: nsMeta})
tableMeta, _ := json.Marshal(tableMetadataInternal{
Name: "t",
Namespace: "ns",
Format: "ICEBERG",
OwnerAccountID: DefaultAccountID,
MetadataVersion: 3,
MetadataLocation: "s3://" + renameTestBucket + "/ns/t/metadata/v3.metadata.json",
})
fs.putEntry(GetNamespacePath(renameTestBucket, "ns"), "t", map[string][]byte{
ExtendedKeyMetadata: tableMeta,
ExtendedKeyMetadataVersion: []byte("3"),
})
// Physical metadata.json and data files live under the table directory.
tablePath := GetTablePath(renameTestBucket, "ns", "t")
fs.putEntry(tablePath, "metadata", nil)
fs.putEntry(tablePath, "data", nil)
fs.putEntry(path.Join(tablePath, "metadata"), "v3.metadata.json", nil)
m := NewManager()
m.SetTrusted(true)
return fs, m
}
func runRename(t *testing.T, m *Manager, fs *memFilerServer, req *RenameTableRequest) error {
t.Helper()
return m.Execute(context.Background(), NewManagerClient(fs.client), "RenameTable", req, nil, "")
}
func runGetTable(t *testing.T, m *Manager, fs *memFilerServer, namespace, name string) (*GetTableResponse, error) {
t.Helper()
resp := &GetTableResponse{}
err := m.Execute(context.Background(), NewManagerClient(fs.client), "GetTable", &GetTableRequest{
TableBucketARN: mustBucketARN(t),
Namespace: []string{namespace},
Name: name,
}, resp, "")
return resp, err
}
func TestRenameTablePreservesData(t *testing.T) {
fs, m := startRenameManager(t)
req := &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "t",
DestNamespace: []string{"ns"},
DestName: "t2",
}
require.NoError(t, runRename(t, m, fs, req))
// The source directory and its metadata.json/data children must survive: rename
// is catalog-only and the destination still points at the original location.
srcPath := GetTablePath(renameTestBucket, "ns", "t")
assert.NotNil(t, fs.getEntry(srcPath, "metadata"), "source metadata dir must survive")
assert.NotNil(t, fs.getEntry(srcPath, "data"), "source data dir must survive")
assert.NotNil(t, fs.getEntry(path.Join(srcPath, "metadata"), "v3.metadata.json"), "metadata.json must survive")
// Source catalog xattrs are dropped so the name stops resolving.
src := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
require.NotNil(t, src, "source directory must remain to hold the data children")
_, hasMeta := src.Extended[ExtendedKeyMetadata]
assert.False(t, hasMeta, "source table-metadata xattr must be removed")
_, err := runGetTable(t, m, fs, "ns", "t")
require.Error(t, err)
var s3Err *S3TablesError
require.ErrorAs(t, err, &s3Err)
assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type)
// The destination resolves to the preserved (original) MetadataLocation.
got, err := runGetTable(t, m, fs, "ns", "t2")
require.NoError(t, err)
assert.Equal(t, "t2", got.Name)
assert.Equal(t, "s3://"+renameTestBucket+"/ns/t/metadata/v3.metadata.json", got.MetadataLocation)
dest := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t2")
require.NotNil(t, dest)
assert.Equal(t, []byte("3"), dest.Extended[ExtendedKeyMetadataVersion])
}
func TestRenameTableSourceMissing(t *testing.T) {
fs, m := startRenameManager(t)
err := runRename(t, m, fs, &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "ghost",
DestNamespace: []string{"ns"},
DestName: "t2",
})
require.Error(t, err)
var s3Err *S3TablesError
require.ErrorAs(t, err, &s3Err)
assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type)
}
func TestRenameTableDestExists(t *testing.T) {
fs, m := startRenameManager(t)
existing, _ := json.Marshal(tableMetadataInternal{Name: "t2", Namespace: "ns", OwnerAccountID: DefaultAccountID})
fs.putEntry(GetNamespacePath(renameTestBucket, "ns"), "t2", map[string][]byte{ExtendedKeyMetadata: existing})
err := runRename(t, m, fs, &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "t",
DestNamespace: []string{"ns"},
DestName: "t2",
})
require.Error(t, err)
var s3Err *S3TablesError
require.ErrorAs(t, err, &s3Err)
assert.Equal(t, ErrCodeTableAlreadyExists, s3Err.Type)
assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched on conflict")
}
func TestRenameTableDestNamespaceMissing(t *testing.T) {
fs, m := startRenameManager(t)
err := runRename(t, m, fs, &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "t",
DestNamespace: []string{"other"},
DestName: "t2",
})
require.Error(t, err)
var s3Err *S3TablesError
require.ErrorAs(t, err, &s3Err)
assert.Equal(t, ErrCodeNoSuchNamespace, s3Err.Type)
assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched")
}
// A principal allowed to rename the source must still be denied when it cannot
// create a table in the destination namespace.
func TestRenameTableDestNamespaceUnauthorized(t *testing.T) {
fs, m := startRenameManager(t)
m.SetTrusted(false)
m.SetDefaultAllow(false)
// "mover" may rename the source table but holds no rights on "dest".
srcPolicy, _ := json.Marshal(map[string]interface{}{
"Version": "2012-10-17",
"Statement": []map[string]interface{}{{
"Effect": "Allow",
"Principal": "mover",
"Action": "s3tables:RenameTable",
"Resource": "*",
}},
})
srcEntry := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
require.NotNil(t, srcEntry)
srcEntry.Extended[ExtendedKeyPolicy] = srcPolicy
destNsMeta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"dest"}, OwnerAccountID: DefaultAccountID})
fs.putEntry(GetTableBucketPath(renameTestBucket), "dest", map[string][]byte{ExtendedKeyMetadata: destNsMeta})
mover := &testIdentity{Name: "mover", Account: &testIdentityAccount{Id: "mover"}}
ctx := s3_constants.SetIdentityInContext(context.Background(), mover)
err := m.Execute(ctx, NewManagerClient(fs.client), "RenameTable", &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "t",
DestNamespace: []string{"dest"},
DestName: "t2",
}, nil, "mover")
require.Error(t, err)
var s3Err *S3TablesError
require.ErrorAs(t, err, &s3Err)
assert.Equal(t, ErrCodeAccessDenied, s3Err.Type)
assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched")
assert.Nil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "dest"), "t2"), "destination must not be written")
}
func TestRenameTableInvalidName(t *testing.T) {
fs, m := startRenameManager(t)
err := runRename(t, m, fs, &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "t",
DestNamespace: []string{"ns"},
DestName: "Bad/Name",
})
require.Error(t, err)
var s3Err *S3TablesError
require.ErrorAs(t, err, &s3Err)
assert.Equal(t, ErrCodeInvalidRequest, s3Err.Type)
}
// A disabled maintenance configuration has to move with the table. Leaving it
// behind re-enables maintenance under the new name, and leaks the old settings
// onto whatever is created at the old one.
func TestRenameTableCarriesMaintenanceConfiguration(t *testing.T) {
fs, m := startRenameManager(t)
src := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
require.NotNil(t, src)
config := []byte(`{"icebergSnapshotManagement":{"status":"disabled"}}`)
status := []byte(`{"icebergCompaction":{"status":"Successful"}}`)
src.Extended[ExtendedKeyMaintenance] = config
src.Extended[ExtendedKeyMaintenanceStatus] = status
require.NoError(t, runRename(t, m, fs, &RenameTableRequest{
TableBucketARN: mustBucketARN(t),
SourceNamespace: []string{"ns"},
SourceName: "t",
DestNamespace: []string{"ns"},
DestName: "t2",
}))
dest := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t2")
require.NotNil(t, dest)
assert.Equal(t, config, dest.Extended[ExtendedKeyMaintenance], "the disable must move with the table")
assert.Equal(t, status, dest.Extended[ExtendedKeyMaintenanceStatus])
// And must not linger on the old name, where a reused name would inherit it.
moved := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
require.NotNil(t, moved)
_, hasConfig := moved.Extended[ExtendedKeyMaintenance]
assert.False(t, hasConfig, "source maintenance configuration must be cleared")
_, hasStatus := moved.Extended[ExtendedKeyMaintenanceStatus]
assert.False(t, hasStatus, "source maintenance status must be cleared")
}
// A maintenance configuration written after the rename copied the source must
// not be deleted by the source cleanup: it would vanish without ever reaching
// the destination.
func TestRenameTableRejectsConcurrentMaintenanceWrite(t *testing.T) {
fs, _ := startRenameManager(t)
src := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
require.NotNil(t, src)
copied := []byte(`{"icebergCompaction":{"status":"enabled"}}`)
src.Extended[ExtendedKeyMaintenance] = copied
// Stand in for the Put that lands between the copy and the cleanup.
landedLate := []byte(`{"icebergSnapshotManagement":{"status":"disabled"}}`)
expected := map[string][]byte{ExtendedKeyMaintenance: copied}
src.Extended[ExtendedKeyMaintenance] = landedLate
h := NewS3TablesHandler()
err := NewManagerClient(fs.client).WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.removeExtendedAttributesIf(context.Background(),
client, GetTablePath(renameTestBucket, "ns", "t"), expected, renamedTableAttributes...)
})
require.Error(t, err, "cleanup must refuse to delete a configuration written after the copy")
assert.ErrorIs(t, err, ErrConcurrentUpdate)
assert.Equal(t, landedLate, src.Extended[ExtendedKeyMaintenance], "the late write must survive")
}
Reference in New Issue View Git Blame Copy Permalink