diff --git a/weed/filer/entry.go b/weed/filer/entry.go index 8a80c7462..796674e06 100644 --- a/weed/filer/entry.go +++ b/weed/filer/entry.go @@ -181,6 +181,16 @@ func (entry *Entry) isS3Entry() bool { return false } +// ApplyStorageTtl stamps the TTL the entry's data was placed with. Remote +// entries never expire locally; the remote storage owns their lifecycle. +func (entry *Entry) ApplyStorageTtl(ttlSec int32) { + if entry.Remote != nil { + ttlSec = 0 + } + entry.TtlSec = ttlSec + entry.ApplyS3ExpiryMetadata() +} + func (entry *Entry) ApplyS3ExpiryMetadata() { if entry.TtlSec == 0 { return diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 07922f2ed..9bc9f9a7a 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -388,19 +388,17 @@ func (fs *FilerServer) ObjectTransactionBatch(ctx context.Context, req *filer_pb // 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. +// none. The returned option carries the same TTL as the entry, so chunks a +// caller still has to place expire with it. func (fs *FilerServer) applyStorageDefaultsToEntry(ctx context.Context, entry *filer.Entry) (*operation.StorageOption, error) { - so, err := fs.detectStorageOption(ctx, string(entry.FullPath), "", "", 0, "", "", "", "") + if entry.Remote != nil { + entry.TtlSec = 0 + } + so, err := fs.detectStorageOption(ctx, string(entry.FullPath), "", "", entry.TtlSec, "", "", "", "") if err != nil { return nil, err } - if entry.Remote != nil { - entry.TtlSec = 0 - } else if entry.TtlSec == 0 { - entry.TtlSec = so.TtlSeconds - } - entry.ApplyS3ExpiryMetadata() + entry.ApplyStorageTtl(so.TtlSeconds) return so, nil } @@ -785,9 +783,9 @@ func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendTo } entry.Chunks = append(entry.GetChunks(), req.Chunks...) - so, err := fs.detectStorageOption(ctx, string(fullpath), "", "", entry.TtlSec, "", "", "", "") + so, err := fs.applyStorageDefaultsToEntry(ctx, entry) if err != nil { - glog.WarningfCtx(ctx, "detectStorageOption: %v", err) + glog.WarningfCtx(ctx, "applyStorageDefaultsToEntry: %v", err) return &filer_pb.AppendToEntryResponse{}, err } entry.Chunks, err = filer.MaybeManifestize(fs.saveAsChunk(ctx, so), entry.GetChunks()) diff --git a/weed/server/filer_server_handlers_copy.go b/weed/server/filer_server_handlers_copy.go index 35a5d1108..2a6364b76 100644 --- a/weed/server/filer_server_handlers_copy.go +++ b/weed/server/filer_server_handlers_copy.go @@ -157,6 +157,21 @@ func (fs *FilerServer) copy(ctx context.Context, w http.ResponseWriter, r *http. return } + // The copy is remote-backed when a data-only copy restores the destination's + // remote pointer, otherwise when the source carries one. Such an entry never + // expires locally, so its cached chunks must not land on a TTL volume either: + // the entry keeps listing them once they are gone, which reads as cached + // rather than remote-only, and nothing re-fetches. + remoteBacked := srcEntry.Remote != nil + if dataOnly && existingDstEntry != nil { + remoteBacked = existingDstEntry.Remote != nil + } + if remoteBacked && so.TtlSeconds != 0 { + withoutTtl := *so + withoutTtl.TtlSeconds = 0 + so = &withoutTtl + } + // Copy the file content and chunks newEntry, err := fs.copyEntry(ctx, srcEntry, finalDstPath, so) if err != nil { @@ -168,6 +183,11 @@ func (fs *FilerServer) copy(ctx context.Context, w http.ResponseWriter, r *http. preserveDestinationMetadataForDataCopy(existingDstEntry, newEntry) } + // The chunks above were placed under the destination's storage option, so the + // entry has to carry its TTL and not the source's (or, for a data-only copy, + // the destination's older one) - otherwise the entry and its data expire apart. + newEntry.ApplyStorageTtl(so.TtlSeconds) + // Pass o_excl = !overwrite so the default copy refuses to replace an // existing destination, while overwrite=true updates the pre-created target. if createErr := fs.filer.CreateEntry(ctx, newEntry, nil, !overwrite, false, nil, false, fs.filer.MaxFilenameLength); createErr != nil { diff --git a/weed/server/filer_server_storage_rule_ttl_test.go b/weed/server/filer_server_storage_rule_ttl_test.go new file mode 100644 index 000000000..33639f156 --- /dev/null +++ b/weed/server/filer_server_storage_rule_ttl_test.go @@ -0,0 +1,165 @@ +package weed_server + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +const ttlRulePrefix = "/buckets/ttl/" + +// addTtlRule gives ttlRulePrefix a 3 minute volume TTL, the fs.configure setting +// whose effect every write path below has to reproduce on the entries it stores. +func addTtlRule(t *testing.T, f *filer.Filer) { + t.Helper() + if err := f.FilerConf.AddLocationConf(&filer_pb.FilerConf_PathConf{ + LocationPrefix: ttlRulePrefix, + Ttl: "3m", + }); err != nil { + t.Fatalf("AddLocationConf: %v", err) + } +} + +// An object written through ObjectTransaction (the routed S3 write path) must +// pick up the path's TTL rule, the same as one written through CreateEntry. +func TestObjectTransactionPutAppliesRuleTtl(t *testing.T) { + store := newRenameTestStore() + store.entries[ttlRulePrefix] = newDirectoryEntry(ttlRulePrefix, 10) + + server := &FilerServer{ + filer: newRenameTestFiler(t, store), + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + addTtlRule(t, server.filer) + + if _, err := server.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: ttlRulePrefix + "obj", + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/ttl", + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, + }, + }}, + }); err != nil { + t.Fatalf("ObjectTransaction: %v", err) + } + + entry, err := store.FindEntry(context.Background(), ttlRulePrefix+"obj") + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if entry.TtlSec != 180 { + t.Errorf("entry TtlSec = %d, want 180", entry.TtlSec) + } +} + +// A copy landing in a TTL path re-uploads its chunks under that path's rule, so +// the entry has to carry the rule's TTL too - whatever TTL the source had. +func TestCopyAppliesRuleTtl(t *testing.T) { + for _, sourceTtlSec := range []int32{0, 600} { + t.Run(fmt.Sprintf("source ttl %d", sourceTtlSec), func(t *testing.T) { + store := newRenameTestStore() + source := newFileEntry("/src.txt", 11) + source.Content = []byte("hello") + source.TtlSec = sourceTtlSec + source.Crtime = time.Now() // a TTL entry older than its TTL reads back as expired + store.entries["/src.txt"] = source + store.entries[ttlRulePrefix] = newDirectoryEntry(ttlRulePrefix, 10) + + server := &FilerServer{ + filer: newRenameTestFiler(t, store), + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + addTtlRule(t, server.filer) + + req := httptest.NewRequest(http.MethodPost, ttlRulePrefix+"dst.txt?cp.from=/src.txt", http.NoBody) + rec := httptest.NewRecorder() + server.PostHandler(rec, req, 0) + if rec.Code != http.StatusNoContent { + t.Fatalf("copy = %d, want %d; body=%q", rec.Code, http.StatusNoContent, rec.Body.String()) + } + + entry, err := store.FindEntry(context.Background(), ttlRulePrefix+"dst.txt") + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if entry.TtlSec != 180 { + t.Errorf("entry TtlSec = %d, want 180", entry.TtlSec) + } + }) + } +} + +// A remote-backed entry copied into a TTL path must not expire locally: the +// remote storage owns its lifecycle, so a local expiry would drop the pointer +// to an object that is still there. +func TestCopyKeepsRemoteEntryUnexpiring(t *testing.T) { + store := newRenameTestStore() + source := newFileEntry("/src.txt", 11) + source.Remote = &filer_pb.RemoteEntry{StorageName: "s3-remote", RemoteSize: 5} + store.entries["/src.txt"] = source + store.entries[ttlRulePrefix] = newDirectoryEntry(ttlRulePrefix, 10) + + server := &FilerServer{ + filer: newRenameTestFiler(t, store), + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + addTtlRule(t, server.filer) + + req := httptest.NewRequest(http.MethodPost, ttlRulePrefix+"dst.txt?cp.from=/src.txt", http.NoBody) + rec := httptest.NewRecorder() + server.PostHandler(rec, req, 0) + if rec.Code != http.StatusNoContent { + t.Fatalf("copy = %d, want %d; body=%q", rec.Code, http.StatusNoContent, rec.Body.String()) + } + + entry, err := store.FindEntry(context.Background(), ttlRulePrefix+"dst.txt") + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if entry.TtlSec != 0 { + t.Errorf("remote entry TtlSec = %d, want 0", entry.TtlSec) + } +} + +// A completed TUS upload uploads its chunks under the target path's rule, so the +// entry it lands has to expire with them. +func TestCompleteTusUploadAppliesRuleTtl(t *testing.T) { + fs, store := newTusTestServer(t, nil) + addTtlRule(t, fs.filer) + + targetPath := ttlRulePrefix + "upload.bin" + seedTusSession(t, fs, store, TusSession{ID: tusTestUploadID, TargetPath: targetPath, Size: 8}) + seedTusChunk(t, fs, store, tusTestUploadID, 0, 8, "3,01637037d6") + + req := tusRequest(http.MethodPatch, "/.tus/.uploads/"+tusTestUploadID, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Content-Type": "application/offset+octet-stream", + "Upload-Offset": "8", + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("PATCH = %d, want %d; body=%q", rec.Code, http.StatusNoContent, rec.Body.String()) + } + + entry, err := store.FindEntry(context.Background(), util.FullPath(targetPath)) + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if entry.TtlSec != 180 { + t.Errorf("entry TtlSec = %d, want 180", entry.TtlSec) + } +} diff --git a/weed/server/filer_server_tus_complete_test.go b/weed/server/filer_server_tus_complete_test.go index 4eaa5db8b..9c30beecb 100644 --- a/weed/server/filer_server_tus_complete_test.go +++ b/weed/server/filer_server_tus_complete_test.go @@ -2,11 +2,13 @@ package weed_server import ( "context" + "errors" "net/http" "net/http/httptest" "strings" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -120,3 +122,36 @@ func TestFilerServer_completeTusUpload_GapRejected(t *testing.T) { t.Fatalf("entry created despite gap") } } + +// TestFilerServer_completeTusUpload_ReadOnlyTarget verifies a read-only prefix +// still refuses the completing upload, before the session's chunks are claimed. +func TestFilerServer_completeTusUpload_ReadOnlyTarget(t *testing.T) { + fs, store := newTusTestServer(t, nil) + if err := fs.filer.FilerConf.AddLocationConf(&filer_pb.FilerConf_PathConf{ + LocationPrefix: "/buckets/frozen/", + ReadOnly: true, + }); err != nil { + t.Fatalf("AddLocationConf: %v", err) + } + + targetPath := "/buckets/frozen/upload.bin" + session := &TusSession{ + ID: tusTestUploadID, + TargetPath: targetPath, + Size: 8, + Offset: 8, + Chunks: []*TusChunkInfo{{Offset: 0, Size: 8, FileId: "3,01637037d6"}}, + } + seedTusSession(t, fs, store, *session) + + err := fs.completeTusUpload(context.Background(), session) + if !errors.Is(err, ErrReadOnly) { + t.Fatalf("completeTusUpload err = %v, want %v", err, ErrReadOnly) + } + if _, findErr := store.FindEntry(context.Background(), util.FullPath(targetPath)); findErr == nil { + t.Fatalf("entry created under a read-only prefix") + } + if consumed, checkErr := fs.isTusSessionConsumed(context.Background(), tusTestUploadID); checkErr != nil || consumed { + t.Fatalf("session consumed = %v (err %v), want false", consumed, checkErr) + } +} diff --git a/weed/server/filer_server_tus_session.go b/weed/server/filer_server_tus_session.go index b5bc6ae1d..e9d32c252 100644 --- a/weed/server/filer_server_tus_session.go +++ b/weed/server/filer_server_tus_session.go @@ -545,11 +545,25 @@ func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSessio // Create the final file entry targetPath := util.FullPath(session.TargetPath) + entry := &filer.Entry{ + FullPath: targetPath, + Attr: filer.Attr{ + Mode: 0644, + Crtime: session.CreatedAt, + Mtime: time.Now(), + Uid: OS_UID, + Gid: OS_GID, + Mime: contentType, + }, + Chunks: fileChunks, + } - // Apply the same read-only / WORM protections the normal write path enforces - // before landing the entry at the client-chosen target path. - if fs.filer.FilerConf.MatchStorageRule(string(targetPath)).ReadOnly { - return fmt.Errorf("%w: %s", ErrReadOnly, targetPath) + // Apply the same storage rule (read-only prefixes, TTL) and WORM protections + // the normal write path enforces before landing the entry at the + // client-chosen target path. + so, err := fs.applyStorageDefaultsToEntry(ctx, entry) + if err != nil { + return err } if wormEnforced, err := fs.wormEnforcedForEntry(ctx, string(targetPath)); err != nil { return fmt.Errorf("check worm: %w", err) @@ -574,21 +588,8 @@ func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSessio return fmt.Errorf("session deleted before completion: %w", err) } - entry := &filer.Entry{ - FullPath: targetPath, - Attr: filer.Attr{ - Mode: 0644, - Crtime: session.CreatedAt, - Mtime: time.Now(), - Uid: OS_UID, - Gid: OS_GID, - Mime: contentType, - }, - Chunks: fileChunks, - } - // Ensure parent directory exists - if err := fs.filer.CreateEntry(ctx, entry, nil, false, false, nil, false, fs.filer.MaxFilenameLength); err != nil { + if err := fs.filer.CreateEntry(ctx, entry, nil, false, false, nil, false, so.MaxFileNameLength); err != nil { return fmt.Errorf("create final file entry: %w", err) }