mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
Chris Lu
2b8349f9e9
A versioned write's only contended mutation is the .versions directory's latest pointer; the version file itself goes to a unique <object>/.versions/<versionId> path. Add a FinalizeVersionedWrite filer op that, under one exclusive lock on the object key, evaluates the precondition against the current latest, stamps the previous latest noncurrent (before the pointer flip so the lifecycle router observes it), then merges the latest pointer / cached metadata into the .versions entry. Key names are passed in, so the filer carries no S3 semantics. Routing and the lock are keyed on the object (objectWriteOwner + lock_key), the same key normal and suspended writes use, so all writes to one object resolve the same owner and serialize on the same lock regardless of versioning state — a versioned and a non-versioned write to the same object can't race on different owners during a versioning-state change. The gateway routes a versioned PutObject's finalize to that owner and tells putToFiler the version path is unique, so it skips the object write lock and the gateway precondition (the op does both atomically). When the owner is unknown or the condition can't reduce to one primitive, it stays on the lock path; on op error it returns InternalError. Versioned COPY, delete markers, suspended versioning, and multipart completion still use the lock and adopt the same op as follow-ups.
96 lines
3.8 KiB
Go
96 lines
3.8 KiB
Go
package weed_server
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
func newFinalizeTestServer(store *renameTestStore) *FilerServer {
|
|
f := newRenameTestFiler(store)
|
|
f.DirBucketsPath = "/buckets"
|
|
return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}
|
|
}
|
|
|
|
func seedVersions(store *renameTestStore, latestFile, latestEtag string) {
|
|
store.entries["/buckets/b/obj/.versions"] = &filer.Entry{
|
|
FullPath: "/buckets/b/obj/.versions",
|
|
Attr: filer.Attr{Inode: 10, Mode: 0755 | (1 << 31)},
|
|
Extended: map[string][]byte{
|
|
s3_constants.ExtLatestVersionFileNameKey: []byte(latestFile),
|
|
s3_constants.ExtLatestVersionETagKey: []byte(latestEtag),
|
|
},
|
|
}
|
|
store.entries["/buckets/b/obj/.versions/"+latestFile] = &filer.Entry{
|
|
FullPath: util.FullPath("/buckets/b/obj/.versions/" + latestFile),
|
|
Attr: filer.Attr{Inode: 11},
|
|
Extended: map[string][]byte{},
|
|
}
|
|
}
|
|
|
|
func finalizeReq(cond *filer_pb.WriteCondition) *filer_pb.FinalizeVersionedWriteRequest {
|
|
return &filer_pb.FinalizeVersionedWriteRequest{
|
|
VersionsDir: "/buckets/b/obj/.versions",
|
|
SetExtended: map[string][]byte{
|
|
s3_constants.ExtLatestVersionFileNameKey: []byte("v2.ver"),
|
|
s3_constants.ExtLatestVersionIdKey: []byte("vid2"),
|
|
},
|
|
DeleteExtended: []string{s3_constants.ExtLatestVersionETagKey},
|
|
PriorLatestKey: s3_constants.ExtLatestVersionFileNameKey,
|
|
NoncurrentSinceKey: s3_constants.ExtNoncurrentSinceNsKey,
|
|
NoncurrentSinceNs: 77777,
|
|
Condition: cond,
|
|
LatestEtagKey: s3_constants.ExtLatestVersionETagKey,
|
|
LatestDeleteMarkerKey: s3_constants.ExtLatestVersionIsDeleteMarker,
|
|
}
|
|
}
|
|
|
|
// The finalize flips the latest pointer and stamps the previously-latest version
|
|
// noncurrent, atomically under the .versions lock.
|
|
func TestFinalizeVersionedWriteFlipAndDemote(t *testing.T) {
|
|
store := newRenameTestStore()
|
|
seedVersions(store, "v1.ver", "etag1")
|
|
fs := newFinalizeTestServer(store)
|
|
|
|
resp, err := fs.FinalizeVersionedWrite(context.Background(), finalizeReq(&filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_NONE}))
|
|
if err != nil {
|
|
t.Fatalf("unexpected err: %v", err)
|
|
}
|
|
if resp.Error != "" {
|
|
t.Fatalf("unexpected resp error: %s", resp.Error)
|
|
}
|
|
|
|
if got := string(store.entries["/buckets/b/obj/.versions"].Extended[s3_constants.ExtLatestVersionFileNameKey]); got != "v2.ver" {
|
|
t.Errorf("latest pointer not flipped: %q", got)
|
|
}
|
|
if got := string(store.entries["/buckets/b/obj/.versions/v1.ver"].Extended[s3_constants.ExtNoncurrentSinceNsKey]); got != "77777" {
|
|
t.Errorf("previous latest not demoted: %q", got)
|
|
}
|
|
}
|
|
|
|
// A failing precondition is reported and nothing is flipped or demoted.
|
|
func TestFinalizeVersionedWritePreconditionFailed(t *testing.T) {
|
|
store := newRenameTestStore()
|
|
seedVersions(store, "v1.ver", "etag1")
|
|
fs := newFinalizeTestServer(store)
|
|
|
|
// IF_NOT_EXISTS against an existing latest must fail.
|
|
resp, err := fs.FinalizeVersionedWrite(context.Background(), finalizeReq(&filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}))
|
|
if err != nil {
|
|
t.Fatalf("unexpected err: %v", err)
|
|
}
|
|
if resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
|
|
t.Fatalf("want PRECONDITION_FAILED, got %v (%q)", resp.ErrorCode, resp.Error)
|
|
}
|
|
if got := string(store.entries["/buckets/b/obj/.versions"].Extended[s3_constants.ExtLatestVersionFileNameKey]); got != "v1.ver" {
|
|
t.Errorf("pointer should be unchanged on precondition failure, got %q", got)
|
|
}
|
|
if _, stamped := store.entries["/buckets/b/obj/.versions/v1.ver"].Extended[s3_constants.ExtNoncurrentSinceNsKey]; stamped {
|
|
t.Errorf("previous latest should not be demoted on precondition failure")
|
|
}
|
|
}
|