diff --git a/weed/s3api/s3_constants/extend_key.go b/weed/s3api/s3_constants/extend_key.go index e085e9bd238..208180167b5 100644 --- a/weed/s3api/s3_constants/extend_key.go +++ b/weed/s3api/s3_constants/extend_key.go @@ -30,6 +30,12 @@ const ( ExtChecksumAlgorithm = "x-seaweedfs-checksum-algorithm" ExtChecksumValue = "x-seaweedfs-checksum-value" + // Lifecycle expiration annotation (use x-seaweedfs- prefix to avoid leaking in generic header loop). + // Stored by the lifecycle walker for not-yet-due objects; read by GET/HEAD handlers to emit + // the x-amz-expiration response header. + // Format: expiry-date="Mon, 02 Jan 2006 15:04:05 GMT", rule-id="" + ExtExpirationKey = "x-seaweedfs-expiration" + // Bucket Policy ExtBucketPolicyKey = "Seaweed-X-Amz-Bucket-Policy" diff --git a/weed/s3api/s3_constants/header.go b/weed/s3api/s3_constants/header.go index 97afcb3920e..cfb980695fb 100644 --- a/weed/s3api/s3_constants/header.go +++ b/weed/s3api/s3_constants/header.go @@ -57,6 +57,9 @@ const ( AmzPartNumberMarker = "X-Amz-Part-Number-Marker" AmzDeleteMarker = "X-Amz-Delete-Marker" + // S3 lifecycle expiration response header + AmzExpiration = "x-amz-expiration" + SeaweedFSUploadId = "X-Seaweedfs-Upload-Id" SeaweedFSMultipartPartsCount = "X-Seaweedfs-Multipart-Parts-Count" SeaweedFSMultipartPartBoundaries = "X-Seaweedfs-Multipart-Part-Boundaries" // JSON: [{part:1,start:0,end:2,etag:"abc"},{part:2,start:2,end:3,etag:"def"}] diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 9871db85069..bf691601527 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -2048,6 +2048,13 @@ func (s3a *S3ApiServer) setResponseHeaders(w http.ResponseWriter, r *http.Reques } } + // Set x-amz-expiration if the lifecycle walker annotated this object. + if entry.Extended != nil { + if v, ok := entry.Extended[s3_constants.ExtExpirationKey]; ok { + w.Header().Set(s3_constants.AmzExpiration, string(v)) + } + } + // Apply S3 passthrough headers from query parameters // AWS S3 supports overriding response headers via query parameters like: // ?response-cache-control=no-cache&response-content-type=application/json diff --git a/weed/s3api/s3lifecycle/bootstrap/walker.go b/weed/s3api/s3lifecycle/bootstrap/walker.go index ac36798728b..61603c7c736 100644 --- a/weed/s3api/s3lifecycle/bootstrap/walker.go +++ b/weed/s3api/s3lifecycle/bootstrap/walker.go @@ -57,6 +57,11 @@ type ListFunc func(ctx context.Context, bucket, start string, cb func(*Entry) er // the caller decides whether to retry from the recorded last_scanned_path. type Dispatcher interface { Delete(ctx context.Context, action *engine.CompiledAction, entry *Entry) error + // Annotate persists the computed expiration date for a not-yet-due object so + // GET/HEAD handlers can return the x-amz-expiration response header without + // re-evaluating lifecycle rules on every request. Errors are non-fatal: a + // missing annotation is a missing header, not a data-loss event. + Annotate(ctx context.Context, bucket string, entry *Entry, expiresAt time.Time, ruleID string) error } // Checkpoint is the resume state. Caller persists it under @@ -137,6 +142,12 @@ func walkEntry(ctx context.Context, snap *engine.Snapshot, bucket string, entry NoncurrentIndex: entry.NoncurrentIndex, Tags: entry.Tags, } + // earliestExpiry tracks the soonest expiration across all matching expiration-kind + // rules that are not yet due. Used to annotate the object after the loop. + var earliestExpiry time.Time + var expiryRuleID string + dispatched := false + for _, key := range keys { action := snap.Action(key) if action == nil { @@ -167,6 +178,15 @@ func walkEntry(ctx context.Context, snap *engine.Snapshot, bucket string, entry } res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, now) if res.Action == s3lifecycle.ActionNone { + // Not yet due: capture the earliest expiration date so we can annotate. + if key.ActionKind == s3lifecycle.ActionKindExpirationDays || + key.ActionKind == s3lifecycle.ActionKindExpirationDate { + dueAt := s3lifecycle.ComputeDueAt(action.Rule, key.ActionKind, info) + if !dueAt.IsZero() && (earliestExpiry.IsZero() || dueAt.Before(earliestExpiry)) { + earliestExpiry = dueAt + expiryRuleID = action.Rule.ID + } + } continue } if err := dispatch.Delete(ctx, action, entry); err != nil { @@ -174,8 +194,18 @@ func walkEntry(ctx context.Context, snap *engine.Snapshot, bucket string, entry bucket, entry.Path, key.ActionKind, err) return err } + dispatched = true stats.S3LifecycleBootstrapDispatchCounter.WithLabelValues(bucket, key.ActionKind.String()).Inc() } + + // Annotate not-yet-expired objects so GET/HEAD can return x-amz-expiration. + // Skip if the object was already dispatched for deletion. + if !dispatched && !earliestExpiry.IsZero() { + if err := dispatch.Annotate(ctx, bucket, entry, earliestExpiry, expiryRuleID); err != nil { + glog.Warningf("lifecycle bootstrap: annotate %s/%s: %v", bucket, entry.Path, err) + // Non-fatal: a missing annotation is just a missing response header. + } + } return nil } diff --git a/weed/s3api/s3lifecycle/bootstrap/walker_test.go b/weed/s3api/s3lifecycle/bootstrap/walker_test.go index 113d20abebf..841cc0f82e5 100644 --- a/weed/s3api/s3lifecycle/bootstrap/walker_test.go +++ b/weed/s3api/s3lifecycle/bootstrap/walker_test.go @@ -14,8 +14,10 @@ import ( // recorder captures dispatched (action, entry) pairs for assertion. type recorder struct { - calls []dispatchCall - err error // when set, every Delete returns this error + calls []dispatchCall + err error // when set, every Delete returns this error + annotateCalls []annotateCall + annotateErr error // when set, every Annotate returns this error } type dispatchCall struct { @@ -23,6 +25,13 @@ type dispatchCall struct { path string } +type annotateCall struct { + bucket string + path string + expiresAt time.Time + ruleID string +} + func (r *recorder) Delete(ctx context.Context, action *engine.CompiledAction, entry *Entry) error { if r.err != nil { return r.err @@ -31,6 +40,19 @@ func (r *recorder) Delete(ctx context.Context, action *engine.CompiledAction, en return nil } +func (r *recorder) Annotate(_ context.Context, bucket string, entry *Entry, expiresAt time.Time, ruleID string) error { + if r.annotateErr != nil { + return r.annotateErr + } + r.annotateCalls = append(r.annotateCalls, annotateCall{ + bucket: bucket, + path: entry.Path, + expiresAt: expiresAt, + ruleID: ruleID, + }) + return nil +} + func mustTime(t *testing.T, s string) time.Time { t.Helper() tm, err := time.Parse(time.RFC3339, s) @@ -401,6 +423,168 @@ func TestWalk_NonMPUDirectorySkipped(t *testing.T) { } } +func TestWalk_NotYetDueExpirationDaysAnnotates(t *testing.T) { + rule := &s3lifecycle.Rule{ + ID: "exp-rule", + Status: s3lifecycle.StatusEnabled, + ExpirationDays: 30, + } + snap := compileEvDriven(t, "bk", rule) + mod := mustTime(t, "2024-01-01T00:00:00Z") + now := mod.Add(s3lifecycle.DaysToDuration(10)) // 10d in, not yet due at 30d + + rec := &recorder{} + _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: "logs/a", IsLatest: true, ModTime: mod}, + }), rec, WalkOptions{Now: now}) + if err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 0 { + t.Fatalf("not-yet-due entry must not dispatch delete, got %v", rec.calls) + } + if len(rec.annotateCalls) != 1 { + t.Fatalf("want 1 Annotate call, got %d", len(rec.annotateCalls)) + } + ac := rec.annotateCalls[0] + if ac.bucket != "bk" { + t.Fatalf("annotate bucket want bk, got %q", ac.bucket) + } + if ac.path != "logs/a" { + t.Fatalf("annotate path want logs/a, got %q", ac.path) + } + if ac.ruleID != "exp-rule" { + t.Fatalf("annotate ruleID want exp-rule, got %q", ac.ruleID) + } + wantExpiry := mod.Add(s3lifecycle.DaysToDuration(30)) + if !ac.expiresAt.Equal(wantExpiry) { + t.Fatalf("annotate expiresAt want %v, got %v", wantExpiry, ac.expiresAt) + } +} + +func TestWalk_NotYetDueDateAnnotates(t *testing.T) { + expDate := mustTime(t, "2099-01-01T00:00:00Z") + rule := &s3lifecycle.Rule{ + ID: "date-rule", + Status: s3lifecycle.StatusEnabled, + ExpirationDate: expDate, + } + snap := compileEvDriven(t, "bk", rule) + now := mustTime(t, "2024-06-01T00:00:00Z") // well before the expiration date + + rec := &recorder{} + _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: "obj/a", IsLatest: true, ModTime: mustTime(t, "2024-01-01T00:00:00Z")}, + }), rec, WalkOptions{Now: now}) + if err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 0 { + t.Fatalf("pre-date entry must not dispatch delete, got %v", rec.calls) + } + if len(rec.annotateCalls) != 1 { + t.Fatalf("want 1 Annotate call, got %d", len(rec.annotateCalls)) + } + if !rec.annotateCalls[0].expiresAt.Equal(expDate) { + t.Fatalf("annotate expiresAt want %v, got %v", expDate, rec.annotateCalls[0].expiresAt) + } + if rec.annotateCalls[0].ruleID != "date-rule" { + t.Fatalf("annotate ruleID want date-rule, got %q", rec.annotateCalls[0].ruleID) + } +} + +func TestWalk_DueActionDoesNotAnnotate(t *testing.T) { + rule := &s3lifecycle.Rule{ + ID: "exp-rule", + Status: s3lifecycle.StatusEnabled, + ExpirationDays: 30, + } + snap := compileEvDriven(t, "bk", rule) + mod := mustTime(t, "2024-01-01T00:00:00Z") + now := mod.Add(s3lifecycle.DaysToDuration(60)) // well past the 30d threshold + + rec := &recorder{} + _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: "obj/a", IsLatest: true, ModTime: mod}, + }), rec, WalkOptions{Now: now}) + if err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 1 || rec.calls[0].path != "obj/a" { + t.Fatalf("due entry must dispatch delete, got %v", rec.calls) + } + if len(rec.annotateCalls) != 0 { + t.Fatalf("dispatched-for-delete entry must not be annotated, got %v", rec.annotateCalls) + } +} + +func TestWalk_EarliestExpirationAnnotated(t *testing.T) { + // Two rules both matching the object; r2 expires sooner. Walker must + // annotate with the earliest expiry date and its rule ID. + mod := mustTime(t, "2024-01-01T00:00:00Z") + r1 := &s3lifecycle.Rule{ + ID: "r1", + Status: s3lifecycle.StatusEnabled, + ExpirationDays: 30, // expires in 30d + } + r2 := &s3lifecycle.Rule{ + ID: "r2", + Status: s3lifecycle.StatusEnabled, + ExpirationDays: 10, // expires in 10d — earliest + } + snap := compileEvDriven(t, "bk", r1, r2) + now := mod.Add(s3lifecycle.DaysToDuration(5)) // 5d in, neither is due + + rec := &recorder{} + _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: "obj/a", IsLatest: true, ModTime: mod}, + }), rec, WalkOptions{Now: now}) + if err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 0 { + t.Fatalf("neither rule is due, no delete expected, got %v", rec.calls) + } + if len(rec.annotateCalls) != 1 { + t.Fatalf("want exactly 1 Annotate call, got %d", len(rec.annotateCalls)) + } + wantExpiry := mod.Add(s3lifecycle.DaysToDuration(10)) + if !rec.annotateCalls[0].expiresAt.Equal(wantExpiry) { + t.Fatalf("want earliest expiry %v, got %v", wantExpiry, rec.annotateCalls[0].expiresAt) + } + if rec.annotateCalls[0].ruleID != "r2" { + t.Fatalf("want ruleID r2 (earliest), got %q", rec.annotateCalls[0].ruleID) + } +} + +func TestWalk_AnnotateErrorIsNonFatal(t *testing.T) { + // Annotate failure must not halt the walk — a missing annotation is + // a missing response header, not a data-loss event. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpirationDays: 30, + } + snap := compileEvDriven(t, "bk", rule) + mod := mustTime(t, "2024-01-01T00:00:00Z") + now := mod.Add(s3lifecycle.DaysToDuration(5)) // not yet due + + rec := &recorder{annotateErr: errors.New("filer write failed")} + cp, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: "obj/a", IsLatest: true, ModTime: mod}, + {Path: "obj/b", IsLatest: true, ModTime: mod}, + }), rec, WalkOptions{Now: now}) + if err != nil { + t.Fatalf("Walk must not fail on Annotate error, got %v", err) + } + if !cp.Completed { + t.Fatalf("walk should complete despite Annotate error") + } + if cp.LastScannedPath != "obj/b" { + t.Fatalf("walk should scan all entries, checkpoint want obj/b, got %q", cp.LastScannedPath) + } +} + func TestWalk_MPUInitDoesNotFireNoncurrent(t *testing.T) { // Same rule covers both AbortMPU and NoncurrentVersionExpiration; the // MPU init record must dispatch only the AbortMPU action. Without the diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go index 4f3f5b75c88..d7a0a8be750 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go @@ -31,6 +31,10 @@ func (d *recordingDispatcher) Delete(_ context.Context, action *engine.CompiledA return d.err } +func (d *recordingDispatcher) Annotate(_ context.Context, _ string, _ *bootstrap.Entry, _ time.Time, _ string) error { + return nil +} + // findShardForPath returns the shard ID for an entry with given path // in given bucket. Helper for tests that want to force entries onto // a known shard. diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go index eaff959b8cd..140ddc48568 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go @@ -3,12 +3,16 @@ package dailyrun import ( "context" "fmt" + "net/http" "time" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/util" "golang.org/x/time/rate" ) @@ -25,6 +29,11 @@ type WalkerDispatcher struct { // daily-run's processMatches uses so the walker and replay paths // can't combine to burst past the cap. nil disables throttling. Limiter *rate.Limiter + // FilerClient and BucketsPath are used by Annotate to write the + // expiration date back to the object's Extended metadata. + // If FilerClient is nil, Annotate is a no-op. + FilerClient filer_pb.SeaweedFilerClient + BucketsPath string } // Compile-time check. @@ -101,3 +110,50 @@ func (d *WalkerDispatcher) Delete(ctx context.Context, action *engine.CompiledAc action.Bucket, objectPath, action.Key.ActionKind, resp.Outcome, resp.Reason) } } + +// Annotate writes the computed expiration date into the object's Extended +// metadata so GET/HEAD handlers can return x-amz-expiration without +// re-evaluating lifecycle rules per request. +// +// Only non-versioned objects (VersionID == "") are annotated; versioned +// objects require path resolution through the .versions/ directory which +// is intentionally deferred. +// +// Errors are logged by the caller but do not halt the walk. +func (d *WalkerDispatcher) Annotate(ctx context.Context, bucket string, entry *bootstrap.Entry, expiresAt time.Time, ruleID string) error { + if d == nil || d.FilerClient == nil { + return nil + } + // Only annotate non-versioned objects for now. + if entry.VersionID != "" { + return nil + } + + objectPath := entry.Path + fullPath := util.NewFullPath(d.BucketsPath+"/"+bucket, objectPath) + dir, name := fullPath.DirAndName() + + resp, err := filer_pb.LookupEntry(ctx, d.FilerClient, &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, + Name: name, + }) + if err != nil || resp == nil || resp.Entry == nil { + return fmt.Errorf("annotate lookup %s/%s: %w", bucket, objectPath, err) + } + + e := resp.Entry + if e.Extended == nil { + e.Extended = make(map[string][]byte) + } + value := fmt.Sprintf("expiry-date=%q, rule-id=%q", + expiresAt.UTC().Format(http.TimeFormat), ruleID) + e.Extended[s3_constants.ExtExpirationKey] = []byte(value) + + if err := filer_pb.UpdateEntry(ctx, d.FilerClient, &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: e, + }); err != nil { + return fmt.Errorf("annotate update %s/%s: %w", bucket, objectPath, err) + } + return nil +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go index f30239394a6..d21b4e097e3 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go @@ -3,16 +3,21 @@ package dailyrun import ( "context" "errors" + "fmt" + "strings" "testing" "time" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/time/rate" + "google.golang.org/grpc" ) // walkerStubClient captures the last LifecycleDeleteRequest so tests @@ -193,3 +198,145 @@ func TestWalkerDispatcher_NilGuardsReturnError(t *testing.T) { nilClient := &WalkerDispatcher{} require.Error(t, nilClient.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})) } + +// annotateFiler is a minimal SeaweedFilerClient stub for Annotate tests. +// Only LookupDirectoryEntry and UpdateEntry are implemented; other methods +// panic if called (inherited from the embedded interface). +type annotateFiler struct { + filer_pb.SeaweedFilerClient + // entries maps "dir\x00name" to the filer Entry to return on lookup. + entries map[string]*filer_pb.Entry + lookupErr error + updateErr error + lastUpdated *filer_pb.UpdateEntryRequest +} + +func (f *annotateFiler) key(dir, name string) string { return dir + "\x00" + name } + +func (f *annotateFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + if f.lookupErr != nil { + return nil, f.lookupErr + } + e := f.entries[f.key(req.Directory, req.Name)] + if e == nil { + return nil, fmt.Errorf("not found") + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil +} + +func (f *annotateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + if f.updateErr != nil { + return nil, f.updateErr + } + f.lastUpdated = req + return &filer_pb.UpdateEntryResponse{}, nil +} + +func TestWalkerDispatcher_AnnotateNilFilerNoOp(t *testing.T) { + d := &WalkerDispatcher{Client: &walkerStubClient{}, FilerClient: nil} + err := d.Annotate(context.Background(), "bkt", &bootstrap.Entry{Path: "obj"}, time.Now(), "r1") + require.NoError(t, err, "nil FilerClient must be a no-op") +} + +func TestWalkerDispatcher_AnnotateVersionedSkipped(t *testing.T) { + filer := &annotateFiler{entries: map[string]*filer_pb.Entry{}} + d := &WalkerDispatcher{ + Client: &walkerStubClient{}, + FilerClient: filer, + BucketsPath: "/buckets", + } + err := d.Annotate(context.Background(), "bkt", &bootstrap.Entry{Path: "obj", VersionID: "v-abc"}, time.Now(), "r1") + require.NoError(t, err, "versioned entry must be skipped silently") + assert.Nil(t, filer.lastUpdated, "no UpdateEntry call expected for versioned entry") +} + +func TestWalkerDispatcher_AnnotateWritesExpiration(t *testing.T) { + expiresAt := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + ruleID := "my-rule" + + entry := &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{FileSize: 100}, + Extended: map[string][]byte{}, + } + filer := &annotateFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/bkt\x00obj": entry, + }, + } + d := &WalkerDispatcher{ + Client: &walkerStubClient{}, + FilerClient: filer, + BucketsPath: "/buckets", + } + + err := d.Annotate(context.Background(), "bkt", &bootstrap.Entry{Path: "obj"}, expiresAt, ruleID) + require.NoError(t, err) + require.NotNil(t, filer.lastUpdated, "UpdateEntry must be called") + + val := string(filer.lastUpdated.Entry.Extended[s3_constants.ExtExpirationKey]) + assert.True(t, strings.Contains(val, "Thu, 01 Jan 2026 00:00:00 GMT"), + "expiry-date must be formatted as HTTP time, got: %s", val) + assert.True(t, strings.Contains(val, ruleID), + "rule-id must be present, got: %s", val) +} + +func TestWalkerDispatcher_AnnotateWritesExpirationSubdirObject(t *testing.T) { + // Objects under subdirectories: "foo/bar" → dir="/buckets/bkt/foo", name="bar" + expiresAt := time.Date(2027, 6, 15, 0, 0, 0, 0, time.UTC) + entry := &filer_pb.Entry{ + Name: "bar", + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{}, + } + filer := &annotateFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/bkt/foo\x00bar": entry, + }, + } + d := &WalkerDispatcher{ + Client: &walkerStubClient{}, + FilerClient: filer, + BucketsPath: "/buckets", + } + + err := d.Annotate(context.Background(), "bkt", &bootstrap.Entry{Path: "foo/bar"}, expiresAt, "rule-x") + require.NoError(t, err) + require.NotNil(t, filer.lastUpdated) + assert.Equal(t, "/buckets/bkt/foo", filer.lastUpdated.Directory) +} + +func TestWalkerDispatcher_AnnotateLookupFailureReturnsError(t *testing.T) { + filer := &annotateFiler{lookupErr: errors.New("lookup failed")} + d := &WalkerDispatcher{ + Client: &walkerStubClient{}, + FilerClient: filer, + BucketsPath: "/buckets", + } + err := d.Annotate(context.Background(), "bkt", &bootstrap.Entry{Path: "obj"}, time.Now(), "r1") + require.Error(t, err) + assert.Contains(t, err.Error(), "lookup failed") + assert.Nil(t, filer.lastUpdated, "UpdateEntry must not be called when lookup fails") +} + +func TestWalkerDispatcher_AnnotateUpdateFailureReturnsError(t *testing.T) { + entry := &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{}, + } + filer := &annotateFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/bkt\x00obj": entry, + }, + updateErr: errors.New("write failed"), + } + d := &WalkerDispatcher{ + Client: &walkerStubClient{}, + FilerClient: filer, + BucketsPath: "/buckets", + } + err := d.Annotate(context.Background(), "bkt", &bootstrap.Entry{Path: "obj"}, time.Now(), "r1") + require.Error(t, err) + assert.Contains(t, err.Error(), "write failed") +}