diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 3a4c11914..e9e71b6c9 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -173,19 +173,11 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr return &filer_pb.CreateEntryResponse{}, fmt.Errorf("CreateEntry cleanupChunks %s %s: %v", req.Directory, req.Entry.Name, err2) } - so, err := fs.detectStorageOption(ctx, string(util.NewFullPath(req.Directory, req.Entry.Name)), "", "", 0, "", "", "", "") - if err != nil { - return nil, err - } newEntry := filer.FromPbEntry(req.Directory, req.Entry) newEntry.Chunks = chunks - // Don't apply TTL to remote entries - they're managed by remote storage - if newEntry.Remote == nil { - if newEntry.TtlSec == 0 { - newEntry.TtlSec = so.TtlSeconds - } - } else { - newEntry.TtlSec = 0 + so, err := fs.applyStorageDefaultsToEntry(ctx, newEntry) + if err != nil { + return nil, err } // Serialize concurrent mutations to the same path on this filer so the @@ -361,6 +353,23 @@ func (fs *FilerServer) ObjectTransactionBatch(ctx context.Context, req *filer_pb return resp, nil } +// applyStorageDefaultsToEntry enforces the path's storage rule (read-only +// prefixes reject the write) and fills in the rule TTL when the entry carries +// none. Remote entries never expire locally; the remote storage owns their +// lifecycle. +func (fs *FilerServer) applyStorageDefaultsToEntry(ctx context.Context, entry *filer.Entry) (*operation.StorageOption, error) { + so, err := fs.detectStorageOption(ctx, string(entry.FullPath), "", "", 0, "", "", "", "") + if err != nil { + return nil, err + } + if entry.Remote != nil { + entry.TtlSec = 0 + } else if entry.TtlSec == 0 { + entry.TtlSec = so.TtlSeconds + } + return so, nil +} + // applyObjectMutation applies a single mutation while the transaction's path // lock is held. PUT entries are expected to be fully prepared by the caller // (chunks resolved); mutations here are metadata-scoped. A DELETE of an absent @@ -373,7 +382,11 @@ func (fs *FilerServer) applyObjectMutation(ctx context.Context, m *filer_pb.Obje return fmt.Errorf("PUT requires an entry") } newEntry := filer.FromPbEntry(m.Directory, m.Entry) - return fs.filer.CreateEntry(ctx, newEntry, nil, false, fromOtherCluster, signatures, false, fs.filer.MaxFilenameLength) + so, err := fs.applyStorageDefaultsToEntry(ctx, newEntry) + if err != nil { + return err + } + return fs.filer.CreateEntry(ctx, newEntry, nil, false, fromOtherCluster, signatures, false, so.MaxFileNameLength) case filer_pb.ObjectMutation_DELETE: fullpath := util.NewFullPath(m.Directory, m.Name) diff --git a/weed/server/filer_grpc_server_object_txn_ttl_test.go b/weed/server/filer_grpc_server_object_txn_ttl_test.go new file mode 100644 index 000000000..bb62690bf --- /dev/null +++ b/weed/server/filer_grpc_server_object_txn_ttl_test.go @@ -0,0 +1,247 @@ +package weed_server + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func mustSetObjectTxnStorageConf(t *testing.T, fs *FilerServer, conf *filer_pb.FilerConf_PathConf) { + t.Helper() + + if err := fs.filer.FilerConf.SetLocationConf(conf); err != nil { + t.Fatalf("SetLocationConf(%q): %v", conf.LocationPrefix, err) + } +} + +func TestObjectTransactionPutAppliesConfiguredTTLAndExpires(t *testing.T) { + fs, store := newTxnTestServer(nil) + + mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ + LocationPrefix: "/buckets/video/", + Ttl: "1h", + }) + + old := time.Now().Add(-2 * time.Hour) + + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/video/expired.mp4", + Mutations: []*filer_pb.ObjectMutation{ + { + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/video", + Entry: &filer_pb.Entry{ + Name: "expired.mp4", + Attributes: &filer_pb.FuseAttributes{ + Crtime: old.Unix(), + Mtime: old.Unix(), + FileMode: 0644, + FileSize: 123, + TtlSec: 0, + }, + }, + }, + }, + }) + if err != nil { + t.Fatalf("ObjectTransaction transport error: %v", err) + } + if resp.Error != "" { + t.Fatalf("ObjectTransaction response error: %q", resp.Error) + } + + entry := store.entries["/buckets/video/expired.mp4"] + if entry == nil { + t.Fatalf("PUT should create /buckets/video/expired.mp4") + } + if got := entry.TtlSec; got != 3600 { + t.Fatalf("TtlSec = %d, want 3600 from fs.configure storage rule", got) + } + + _, err = fs.filer.FindEntry(context.Background(), util.FullPath("/buckets/video/expired.mp4")) + if err != filer_pb.ErrNotFound { + t.Fatalf("expired entry lookup error = %v, want %v", err, filer_pb.ErrNotFound) + } + + if _, ok := store.entries["/buckets/video/expired.mp4"]; ok { + t.Fatalf("expired entry should be deleted from metadata store") + } +} + +func TestObjectTransactionPutPreservesExplicitTTL(t *testing.T) { + fs, store := newTxnTestServer(nil) + + mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ + LocationPrefix: "/buckets/video/", + Ttl: "1h", + }) + + now := time.Unix(1700000000, 0) + + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/video/lifecycle.mp4", + Mutations: []*filer_pb.ObjectMutation{ + { + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/video", + Entry: &filer_pb.Entry{ + Name: "lifecycle.mp4", + Attributes: &filer_pb.FuseAttributes{ + Crtime: now.Unix(), + Mtime: now.Unix(), + FileMode: 0644, + FileSize: 123, + TtlSec: 7200, + }, + }, + }, + }, + }) + if err != nil { + t.Fatalf("ObjectTransaction transport error: %v", err) + } + if resp.Error != "" { + t.Fatalf("ObjectTransaction response error: %q", resp.Error) + } + + entry := store.entries["/buckets/video/lifecycle.mp4"] + if entry == nil { + t.Fatalf("PUT should create /buckets/video/lifecycle.mp4") + } + if got := entry.TtlSec; got != 7200 { + t.Fatalf("TtlSec = %d, want explicit TTL 7200 to win over fs.configure", got) + } +} + +func TestObjectTransactionPutClearsTTLForRemoteEntry(t *testing.T) { + fs, store := newTxnTestServer(nil) + + mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ + LocationPrefix: "/buckets/video/", + Ttl: "1h", + }) + + now := time.Unix(1700000000, 0) + + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/video/remote", + Mutations: []*filer_pb.ObjectMutation{ + { + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/video", + Entry: &filer_pb.Entry{ + Name: "remote", + Attributes: &filer_pb.FuseAttributes{ + Crtime: now.Unix(), + Mtime: now.Unix(), + FileMode: 0644, + FileSize: 123, + TtlSec: 7200, + }, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 123}, + }, + }, + }, + }) + if err != nil { + t.Fatalf("ObjectTransaction transport error: %v", err) + } + if resp.Error != "" { + t.Fatalf("ObjectTransaction response error: %q", resp.Error) + } + + entry := store.entries["/buckets/video/remote"] + if entry == nil { + t.Fatalf("remote entry should be created") + } + if got := entry.TtlSec; got != 0 { + t.Fatalf("remote entry TtlSec = %d, want 0", got) + } +} + +func TestObjectTransactionPutHonorsMaxFileNameLengthRule(t *testing.T) { + fs, store := newTxnTestServer(nil) + + mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ + LocationPrefix: "/buckets/video/", + MaxFileNameLength: 8, + }) + + now := time.Unix(1700000000, 0) + + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/video/toolong.mp4", + Mutations: []*filer_pb.ObjectMutation{ + { + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/video", + Entry: &filer_pb.Entry{ + Name: "toolong.mp4", + Attributes: &filer_pb.FuseAttributes{ + Crtime: now.Unix(), + Mtime: now.Unix(), + FileMode: 0644, + FileSize: 123, + }, + }, + }, + }, + }) + if err != nil { + t.Fatalf("ObjectTransaction transport error: %v", err) + } + if resp.Error == "" { + t.Fatalf("ObjectTransaction should reject a name longer than the rule's max_file_name_length") + } + + if _, ok := store.entries["/buckets/video/toolong.mp4"]; ok { + t.Fatalf("over-long name must not create the entry") + } +} + +func TestObjectTransactionPutFailsOnReadOnlyStorageConfig(t *testing.T) { + // An explicit TTL must not bypass the read-only rule. + for _, ttlSec := range []int32{0, 7200} { + fs, store := newTxnTestServer(nil) + + mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ + LocationPrefix: "/buckets/video/", + ReadOnly: true, + }) + + now := time.Unix(1700000000, 0) + + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/video/readonly.mp4", + Mutations: []*filer_pb.ObjectMutation{ + { + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/video", + Entry: &filer_pb.Entry{ + Name: "readonly.mp4", + Attributes: &filer_pb.FuseAttributes{ + Crtime: now.Unix(), + Mtime: now.Unix(), + FileMode: 0644, + FileSize: 123, + TtlSec: ttlSec, + }, + }, + }, + }, + }) + if err != nil { + t.Fatalf("TtlSec=%d: ObjectTransaction transport error: %v", ttlSec, err) + } + if resp.Error == "" { + t.Fatalf("TtlSec=%d: ObjectTransaction should report read-only storage config error", ttlSec) + } + + if _, ok := store.entries["/buckets/video/readonly.mp4"]; ok { + t.Fatalf("TtlSec=%d: read-only storage config must not create the entry", ttlSec) + } + } +}