mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-06-13 23:36:45 +03:00
test(s3/lifecycle): integration coverage for versioning + filters (#9415)
* test(s3/lifecycle): integration coverage for versioning + filters
First integration-test bundle building on the existing single-test
backdating harness. Each scenario follows the same shape: create
bucket, set lifecycle, PUT object, backdate mtime via filer
UpdateEntry, run the shell command for one shard sweep, assert
S3-side state.
Five new tests:
- TestLifecycleVersionedBucketCreatesDeleteMarker: Expiration on a
versioned bucket must produce a delete marker (latest after worker
runs is a marker) AND keep the original version directly addressable
by versionId. ListObjectVersions confirms IsLatest=true on the
marker.
- TestLifecycleNoncurrentVersionExpiration: NoncurrentVersionExpiration
fires only on demoted versions. PUT v1, PUT v2 (so v1 → noncurrent),
backdate v1, run worker. v1 must be gone, v2 still current.
- TestLifecycleExpiredDeleteMarkerCleanup: combined rule (noncurrent +
expired-delete-marker) cleans up a sole-survivor marker. PUT v1,
DELETE (creates marker), backdate both, run worker. Every version
AND marker must be gone for the key.
- TestLifecycleDisabledRuleSkipsObject: rule with Status=Disabled
must not produce dispatches even on a backdated match. Negative
test for the engine's enabled-status gate.
- TestLifecycleTagFilter: rule with And{Prefix, Tag} only matches
objects carrying the tag. Two backdated objects (one tagged, one
not) — only the tagged one is removed.
Helpers extracted to keep each test focused: putVersioningEnabled,
putNoncurrentExpirationLifecycle, putExpiredDeleteMarkerLifecycle,
backdateVersionedMtime (ages a specific .versions/v_<id> entry),
runLifecycleShard (one-shot shell invocation with FATAL guard).
* test(s3/lifecycle): tighten noncurrent expiration diagnostics
Local run showed TestLifecycleNoncurrentVersionExpiration failing
with a bare 404 on HEAD(latest), not enough to tell whether v2 was
deleted, the bare-key pointer was removed, or a delete marker was
synthesized. Strengthen the test to:
- HEAD by versionId=v2 first, so we pin "v2 file still on disk"
separately from "the latest pointer resolves to v2"
- on HEAD(latest) failure, log ListObjectVersions output (versions +
markers, with IsLatest) so the next failure shows which side the
bug is on rather than just NotFound
* test(s3/lifecycle): integration coverage for AbortIncompleteMultipartUpload
Exercises the lifecycleAbortMPU handler path that the prefix-based
expiration tests can't reach — routing keys off of .uploads/<id>/
directory events, not regular object events, and the dispatcher uses
a different RPC path (rm on the .uploads/<id>/ folder).
Setup: AbortIncompleteMultipartUpload rule with DaysAfterInitiation=1,
CreateMultipartUpload, UploadPart (so the directory carries the
right shape), backdate the .uploads/<uploadID>/ directory entry 30
days, run the worker. The upload must drop out of
ListMultipartUploads.
Helpers added: putAbortMPULifecycle, backdateUploadDir.
* test(s3/lifecycle): integration coverage for NewerNoncurrentVersions
NewerNoncurrentVersions=N keeps the N most recent noncurrent versions
and expires the rest. Distinct from per-version NoncurrentDays —
depends on per-version rank, not just per-version age — and routes
through routePointerTransition's "needs full expansion" path.
Setup: PUT v1, v2, v3, v4 on a versioned bucket (v4 current; v1-v3
noncurrent), backdate v1+v2+v3 so all satisfy the NoncurrentDays>=1
floor, run the worker. Expect v1+v2 expired (older noncurrent),
v3 (newest noncurrent within keep=1) and v4 (current) preserved.
Helper added: putNewerNoncurrentLifecycle.
* test(s3/lifecycle): integration coverage for suspended-versioning Expiration
Suspended versioning takes a distinct code path in lifecycleDispatch:
the VersioningSuspended branch first deletes the null version (via
deleteSpecificObjectVersion(versionId="null")) and then writes a
fresh delete marker on top. Other branches (Enabled → only writes a
marker; Off → straight rm) miss this two-step.
Setup: enable versioning, PUT v1 (real versionId), suspend
versioning, PUT again (creates the null version, demotes v1 to
noncurrent), set the Expiration rule, backdate the null at the
bare path. Expect: latest is now a fresh delete marker, the
"null" version is gone from ListObjectVersions, and v1 (noncurrent
under Enabled) still addressable directly — suspended Expiration
must only touch the null, not other versions.
Helper added: putVersioningSuspended.
* test(s3/lifecycle): integration coverage for multi-bucket sweep
A single shell-driven shard sweep must process every bucket carrying
lifecycle config, not just the first one alphabetically. Pinned
because the scheduler iterates the buckets directory and a regression
that returns early after the first match would silently disable
lifecycle for every later bucket.
Two buckets, each with their own prefix-expiration rule and a
backdated object. Both must be expired after the same sweep.
* test(s3/lifecycle): integration coverage for ObjectSizeGreaterThan filter
ObjectSizeGreaterThan is a strict > gate (filterAllows uses
ev.Size <= rule.FilterSizeGreaterThan to reject). Pinned at the
boundary: an object whose size equals the threshold must remain;
only an object strictly larger expires. Catches a > vs >= flip.
Two backdated objects on the same prefix, sizes 100 and 150 with
threshold=100 — boundary survives, larger expires.
* test(s3/lifecycle): scrub bucket lifecycle config + versions on cleanup
Tests share one weed mini server. Two pollution modes were producing
order-dependent failures:
- A later test's shard sweep would still load the prior test's
lifecycle config (the worker reads every bucket's XML from filer
state, and DeleteBucket alone doesn't drop lifecycle config
cleanly on this codebase).
- Versioned-bucket tests left versions + delete markers behind that
ListObjectsV2 can't see, so the existing best-effort empty-then-
delete didn't actually empty those buckets.
- The AbortMPU test intentionally leaves an in-flight upload; without
an explicit AbortMultipartUpload the bucket DELETE hits NotEmpty.
Cleanup now runs DeleteBucketLifecycle, ListObjectVersions →
DeleteObject(versionId), ListObjectsV2 → DeleteObject (catches what
ListObjectVersions missed), ListMultipartUploads → AbortMultipartUpload,
then DeleteBucket. Best-effort throughout so a half-torn-down bucket
doesn't fail the cleanup chain.
* test(s3/lifecycle): backdate both versions for NoncurrentDays clock
Per codex review: NoncurrentDays is clocked from the SUCCESSOR
version's mtime (when the displaced version became noncurrent), not
from the displaced version's own mtime. Backdating only v1 left the
clock (v2's mtime) at "now" and the rule never fired — the test was
wrong, not the production path.
Backdate v1=31d and v2=30d so v1 sits past the 1-day threshold
relative to v2, the noncurrent rule fires, and v2 stays current.
* test(s3/lifecycle): assert specific NotFound on multi-bucket deletion
Per codex review: TestLifecycleMultipleBucketsInOneSweep treated any
HeadObject error as "deleted", which lets a transport failure or
dead endpoint mask a real bug. Recognize NoSuchKey/NotFound/HTTP-404
specifically via a small isS3NotFound helper so the assertion
actually proves deletion happened, not just that the call broke.
* test(s3/lifecycle): gofmt size-filter test
* test(s3/lifecycle): integration coverage for Object Lock skip
Object Lock retention must override the lifecycle rule. The handler's
enforceObjectLockProtections check (s3api_internal_lifecycle.go:47)
returns an error when retention is active; the dispatcher then
classifies the outcome as SKIPPED_OBJECT_LOCK and the object stays.
No existing integration test reaches that outcome.
Setup: bucket created with ObjectLockEnabledForBucket=true, expiration
rule on prefix "lock/", two backdated objects under the same prefix —
one with GOVERNANCE retention until 1h from now, one without. After
the worker runs, the unlocked object expires (positive control); the
locked one survives.
Custom cleanup uses BypassGovernanceRetention so the test can drop
the locked version when the test finishes — otherwise the retention
window keeps the bucket from being deleted.
* test(s3/lifecycle): integration coverage for config update between sweeps
An operator changes the lifecycle rule between two shell-driven
sweeps. The second sweep must respect the NEW rule, not a cached
copy of the old one. Each runLifecycleShard invocation spawns a
fresh weed shell subprocess, so cached engine state from a previous
sweep doesn't persist — but a regression that caches rules across
PutBucketLifecycleConfiguration calls within the S3 server itself
would still surface here.
Sweep 1: rule prefix="first/", PUT + backdate firstKey, run worker
→ firstKey expires.
Update rule to prefix="second/", PUT + backdate secondKey AND a
new key under the OLD prefix ("first/post-update.txt"). Sweep 2
must expire only the second-prefix object; the post-update old-
prefix one must survive — config replacement, not merge.
* test(s3/lifecycle): integration coverage for ExpirationDate (past)
Rules with Expiration{Date: <past>} route through ScanAtDate in the
engine (decideMode's ActionKindExpirationDate case) — a separate
compile + dispatch branch from the EventDriven delay-group path the
Days-based tests exercise.
Past date + in-prefix object → must expire. Out-of-prefix object →
must remain. Object also backdated as defense-in-depth so the
assertion doesn't depend on whether the dispatcher consults
MinTriggerAge for date kinds.
* test(s3/lifecycle): integration coverage for bootstrap walk on existing objects
Production scenario: operator enables lifecycle on a bucket that
already holds objects from before the policy. The worker must
discover them via the bootstrap walk (BucketBootstrapper) — there
were no meta-log events to observe because the objects predate the
rule. Without the bootstrap path, only NEW writes would ever match.
Setup: PUT 5 objects (no lifecycle config yet) + 1 out-of-prefix
survivor, backdate all, THEN set the Expiration rule, run the
worker. Every in-prefix pre-existing object must be expired; the
out-of-prefix one must remain.
* test(s3/lifecycle): integration coverage for DeleteBucketLifecycle stops dispatching
Operator UX: after DeleteBucketLifecycle, the worker must observe the
removal on the next sweep and stop expiring objects under the now-gone
rule. A regression that caches old configs across
PutBucketLifecycleConfiguration → DeleteBucketLifecycle would keep
silently dropping objects.
Setup: positive control (rule active, backdated obj expires) →
DeleteBucketLifecycle → PUT + backdate a fresh object → second
sweep. The fresh object must remain.
* test(s3/lifecycle): integration coverage for empty bucket sweep no-op
A bucket carrying lifecycle config but no objects must produce a
successful sweep — no hangs, no errors, no dispatches. Pinned
because the bootstrap walker iterates bucket directories, and an
empty directory is a corner of that traversal that's easy to break
(slice-bounds bug on the first listing returning zero entries).
Asserts: worker logs "loaded lifecycle for" and "shards 0-15
complete", no FATAL output, bucket still exists after the sweep.
* test(s3/lifecycle): fix Object Lock backdate path + skip unwired ScanAtDate
ObjectLock: enabling Object Lock on a bucket implicitly enables
versioning, so PUT objects land at .versions/v_<id>, not at the bare
key. The test was calling backdateMtime (bare path) and failing in
the helper with "filer: no entry is found". Switch to
backdateVersionedMtime with the versionId returned by PutObject.
ExpirationDate: ScanAtDate dispatch path isn't wired to the run-shard
shell command yet — the bootstrap walker explicitly skips actions in
ModeScanAtDate (walker.go:141 says "SCAN_AT_DATE runs its own date-
triggered bootstrap" but no such bootstrap exists in the scheduler or
shell). Skip with a t.Skip + explanation so the test activates the
moment the date-triggered path lands.
* fix(s3/lifecycle): wire ExpirationDate dispatch through bootstrap walker
The walker explicitly skipped ModeScanAtDate actions on the comment
"SCAN_AT_DATE runs its own date-triggered bootstrap" — but no such
bootstrap exists in the scheduler or shell layer. The result: rules
with Expiration{Date: ...} compiled correctly, populated the
snapshot's dateActions map, and were never dispatched.
ExpirationDate is silently a no-op in production.
EvaluateAction already handles ActionKindExpirationDate correctly
(rejects when now.Before(rule.ExpirationDate), otherwise emits
ActionDeleteObject). The walker just needed to fall through instead
of skipping. Pre-date walks become no-ops via EvaluateAction's date
check; post-date walks expire eligible objects.
Un-skip TestLifecycleExpirationDateInThePast — it now exercises the
fixed path end-to-end.
* test(s3/lifecycle): integration coverage for multiple rules per bucket
A single bucket carries two independent Expiration rules with disjoint
prefix filters and different Days thresholds. Each rule must fire
only on its prefix; objects outside both prefixes must survive.
Pinned because Compile builds one CompiledAction per rule per kind
all sharing the same bucket index — a bug that lets one rule's
prefix or threshold leak into another (e.g. last-write-wins on a
shared map) would silently expire wrong objects.
Setup: rule A with prefix=logs/ Days=1, rule B with prefix=tmp/
Days=7. Three backdated objects: logs/access.log, tmp/scratch.bin,
data/keep.bin. After the worker runs, logs/ + tmp/ are gone;
data/ — outside both rule prefixes — survives.
* fix(s3/lifecycle): mark ScanAtDate actions active in Compile
Two layers were silently filtering ScanAtDate actions out of routing:
the walker's mode skip (fixed in e785f59d6) and Compile only marking
ModeEventDriven actions active. MatchPath / MatchOriginalWrite both
require IsActive() to emit a key, so a ScanAtDate action that's never
marked active never reaches a dispatch path even after the walker
falls through.
ScanAtDate's only dispatch path is the bootstrap walk's MatchPath
call — there's no bootstrap-completion rendezvous to wait on. Make
the active flag include ModeScanAtDate alongside the
EventDriven+BootstrapComplete combination.
ExpirationDate-based rules now actually fire end-to-end. The
TestLifecycleExpirationDateInThePast integration test exercises this.
* fix(s3/lifecycle): route date kinds via ComputeDueAt
ExpirationDate has MinTriggerAge=0, so router computed
dueTime = info.ModTime + 0 = info.ModTime. For a backdated entry
that mtime is BEFORE rule.ExpirationDate, so EvaluateAction's
now.Before(rule.ExpirationDate) check returned ActionNone and the
date rule never fired through the event-driven path.
ComputeDueAt already knows the per-kind shape — rule.ExpirationDate
for date kinds, ModTime+Days for the rest — so use it as the
single source of truth for dueTime in Route's main loop.
* test(s3/lifecycle): pin bootstrap walker date dispatch
The original TestWalk_DateActionsSkipped pinned the pre-e785f59d6
behavior that the regular walker skipped ExpirationDate. That
walker was rewired to fire date rules whose date has passed (the
SCAN_AT_DATE bootstrap was never wired); update the test to match.
Split into two: post-date entries dispatch, pre-date entries don't.
* test(s3/lifecycle): drop unused putExpiredDeleteMarkerLifecycle
The helper was never called — TestLifecycleExpiredDeleteMarkerCleanup
constructs a combined noncurrent + expired-marker rule inline, which
the helper doesn't cover. The blank-assignment workaround was just
hiding dead code; remove both.
* test(s3/lifecycle): tighten HeadObject termination check to typed not-found
Generic err != nil also passes on transport/auth/timeouts, letting
the test go green without proving the lifecycle action actually
fired. Switch the three Eventuallyf HeadObject predicates to
isS3NotFound, matching the pattern already in the multi-bucket and
expiration-date tests.
* test(s3/lifecycle): guard ListObjectVersions diagnostic against nil
When ListObjectVersions errors, listOut is nil and the diagnostic
log path panics on listOut.Versions before the real assertion fires.
Branch on (listErr != nil || listOut == nil) so the failure log is
robust whatever ListObjectVersions returned.
This commit is contained in:
Chris Lu
GitHub
@@ -0,0 +1,79 @@
|
||||
// Bootstrap-walk integration scenario: lifecycle config added AFTER the
|
||||
// objects exist must still expire them on the first sweep.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleBootstrapWalkOnExistingObjects: PUT a batch of objects
|
||||
// FIRST, backdate them, THEN configure the lifecycle rule, THEN run
|
||||
// the worker. The worker must discover the pre-existing objects via
|
||||
// the bootstrap walk (BucketBootstrapper) — there were no meta-log
|
||||
// events the reader could have observed because the objects predate
|
||||
// the rule.
|
||||
//
|
||||
// Production scenario: an operator enables lifecycle on a bucket that
|
||||
// already holds a million objects from before the policy. Without a
|
||||
// bootstrap walk, only NEW writes would ever match; the existing
|
||||
// content would never expire.
|
||||
func TestLifecycleBootstrapWalkOnExistingObjects(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("bootstrap")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
|
||||
// 5 objects so the test exercises the walker iterating multiple
|
||||
// entries in the same bucket directory. PUT FIRST, before any
|
||||
// lifecycle config is set, so the meta-log doesn't carry events
|
||||
// that match any rule.
|
||||
const total = 5
|
||||
keys := make([]string, total)
|
||||
for i := 0; i < total; i++ {
|
||||
keys[i] = fmt.Sprintf("preexisting/obj-%02d.txt", i)
|
||||
putObject(t, c, bucket, keys[i], "content")
|
||||
backdateMtime(t, fc, bucket, keys[i], 30)
|
||||
}
|
||||
|
||||
// One out-of-prefix object that must NOT be expired. Same
|
||||
// pre-existing PUT timing.
|
||||
const survivor = "other/keep.txt"
|
||||
putObject(t, c, bucket, survivor, "keep")
|
||||
backdateMtime(t, fc, bucket, survivor, 30)
|
||||
|
||||
// NOW set the rule. Up to this point the worker (if it were
|
||||
// running event-driven only) would have nothing matching to do.
|
||||
putExpirationLifecycle(t, c, bucket, "preexisting/", 1)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// Every pre-existing in-prefix object must be expired by the
|
||||
// bootstrap walk's discovery + dispatch.
|
||||
for _, k := range keys {
|
||||
k := k
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(k),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond,
|
||||
"pre-existing %s/%s must be discovered + expired by bootstrap walk", bucket, k)
|
||||
}
|
||||
|
||||
// Out-of-prefix object stays — pin that the bootstrap walk
|
||||
// doesn't ignore the rule's prefix filter.
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(survivor),
|
||||
})
|
||||
require.NoError(t, err, "out-of-prefix object must survive the bootstrap walk")
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
// Lifecycle config update across sweeps.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleConfigUpdateBetweenSweeps: an operator changes the
|
||||
// lifecycle rule between two shell-driven sweeps. The second sweep
|
||||
// must respect the NEW rule, not a cached version of the old one.
|
||||
//
|
||||
// Each `runLifecycleShard` invocation spawns a fresh `weed shell`
|
||||
// subprocess, so cached engine state from a previous sweep doesn't
|
||||
// persist across runs. This test pins that the freshly-loaded config
|
||||
// actually changes routing — under the new prefix only matching
|
||||
// objects expire, even if there are still backdated objects sitting
|
||||
// under the old prefix.
|
||||
func TestLifecycleConfigUpdateBetweenSweeps(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("config-update")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
|
||||
// Sweep 1: rule expires anything under "first/".
|
||||
putExpirationLifecycle(t, c, bucket, "first/", 1)
|
||||
|
||||
const firstKey = "first/initial.txt"
|
||||
putObject(t, c, bucket, firstKey, "first")
|
||||
backdateMtime(t, fc, bucket, firstKey, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("sweep 1 output:\n%s", out)
|
||||
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(firstKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "sweep 1 must expire %s", firstKey)
|
||||
|
||||
// Update the rule to a different prefix. The old "first/" prefix
|
||||
// is no longer covered by any rule; objects under it must NOT be
|
||||
// expired by sweep 2 even when backdated.
|
||||
putExpirationLifecycle(t, c, bucket, "second/", 1)
|
||||
|
||||
const secondKey = "second/new.txt"
|
||||
const oldPrefixKey = "first/post-update.txt"
|
||||
putObject(t, c, bucket, secondKey, "second")
|
||||
putObject(t, c, bucket, oldPrefixKey, "stale rule")
|
||||
backdateMtime(t, fc, bucket, secondKey, 30)
|
||||
backdateMtime(t, fc, bucket, oldPrefixKey, 30)
|
||||
|
||||
out = runLifecycleShard(t)
|
||||
t.Logf("sweep 2 output:\n%s", out)
|
||||
|
||||
// Sweep 2 expires the new-prefix object.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(secondKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "sweep 2 must expire %s under the new rule", secondKey)
|
||||
|
||||
// Sweep 2 must NOT expire the old-prefix object — the rule was
|
||||
// replaced, not merged. A regression that caches old rules across
|
||||
// PutBucketLifecycleConfiguration calls would fail here.
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(oldPrefixKey),
|
||||
})
|
||||
require.NoError(t, err, "old-prefix object must survive after rule update — config replacement, not merge")
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
// Operator-removes-lifecycle-config integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleDeleteBucketLifecycleStopsDispatching: when an operator
|
||||
// runs DeleteBucketLifecycle, subsequent sweeps must NOT continue
|
||||
// expiring objects under the now-removed rule. The worker reads each
|
||||
// bucket's XML config from filer state on every run; if it cached the
|
||||
// previous config across PutBucketLifecycleConfiguration → DeleteBucket
|
||||
// Lifecycle the bucket would keep losing objects.
|
||||
//
|
||||
// Setup: positive control (rule active, backdated obj expires), then
|
||||
// DeleteBucketLifecycle, then PUT + backdate a fresh object, run the
|
||||
// worker again. The fresh object must remain — no rule, no dispatch.
|
||||
func TestLifecycleDeleteBucketLifecycleStopsDispatching(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("delete-config")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putExpirationLifecycle(t, c, bucket, "x/", 1)
|
||||
|
||||
const firstKey = "x/initial.txt"
|
||||
putObject(t, c, bucket, firstKey, "first")
|
||||
backdateMtime(t, fc, bucket, firstKey, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("sweep 1 output (rule active):\n%s", out)
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(firstKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "positive control: %s should expire while rule is active", firstKey)
|
||||
|
||||
// Now remove the rule. The worker must observe the removal on the
|
||||
// next sweep.
|
||||
_, err := c.DeleteBucketLifecycle(context.Background(), &s3.DeleteBucketLifecycleInput{
|
||||
Bucket: aws.String(bucket),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const survivorKey = "x/post-delete.txt"
|
||||
putObject(t, c, bucket, survivorKey, "should-stay")
|
||||
backdateMtime(t, fc, bucket, survivorKey, 30)
|
||||
|
||||
out = runLifecycleShard(t)
|
||||
t.Logf("sweep 2 output (rule deleted):\n%s", out)
|
||||
|
||||
// Wait the same window; survivor MUST still exist after the worker
|
||||
// has had every opportunity to (incorrectly) act on it.
|
||||
time.Sleep(2 * time.Second)
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(survivorKey),
|
||||
})
|
||||
require.NoError(t, err, "after DeleteBucketLifecycle, sweeps must not expire %s", survivorKey)
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
// Empty bucket sweep — no-op safety integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleEmptyBucketSweepIsNoOp: a bucket carrying lifecycle
|
||||
// config but no objects must produce a successful sweep — no hangs,
|
||||
// no errors, no dispatches. Pinned because the bootstrap walker
|
||||
// iterates bucket directories, and an empty directory is a corner of
|
||||
// that traversal that's easy to break (e.g. a slice-bounds bug on
|
||||
// the first listing returning zero entries).
|
||||
func TestLifecycleEmptyBucketSweepIsNoOp(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
_, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("empty")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putExpirationLifecycle(t, c, bucket, "anywhere/", 1)
|
||||
|
||||
// No PUTs. The bucket is empty.
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("sweep output:\n%s", out)
|
||||
|
||||
// Sanity: the worker must report it loaded the bucket's config and
|
||||
// completed without errors. The runLifecycleShard helper already
|
||||
// fails on FATAL; pin the success markers here too so a regression
|
||||
// that produces a half-shaped output (no completion) is caught.
|
||||
require.Contains(t, out, "loaded lifecycle for", "worker must report config load")
|
||||
require.Contains(t, out, "shards 0-15 complete", "worker must complete all shards")
|
||||
require.False(t, strings.Contains(out, "FATAL"), "worker must not produce FATAL output")
|
||||
|
||||
// Bucket still exists after the sweep — the worker doesn't drop
|
||||
// empty buckets as a side effect.
|
||||
_, err := c.HeadBucket(context.Background(), &s3.HeadBucketInput{Bucket: aws.String(bucket)})
|
||||
require.NoError(t, err, "empty bucket must still exist after sweep")
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
// ExpirationDate (date-based, not Days) integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleExpirationDateInThePast: an Expiration{Date: <past>} rule
|
||||
// routes through the engine's ScanAtDate mode rather than the
|
||||
// EventDriven delay-group path. The worker's dispatcher must still fire
|
||||
// when the date has already passed at the time the worker runs. Pinning
|
||||
// this path because most tests use Days-based rules; ScanAtDate is a
|
||||
// separate compile + dispatch branch (engine.decideMode case
|
||||
// ActionKindExpirationDate) that wouldn't be exercised otherwise.
|
||||
func TestLifecycleExpirationDateInThePast(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("expdate")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
|
||||
// Date in the past: AWS rejects ExpirationDate in the future from
|
||||
// being processed early but a past date is the natural integration
|
||||
// test — every object hit by the rule is immediately eligible.
|
||||
pastDate := time.Now().Add(-7 * 24 * time.Hour).UTC().Truncate(24 * time.Hour)
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{{
|
||||
ID: aws.String("expire-by-date"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String("d/")},
|
||||
Expiration: &types.LifecycleExpiration{
|
||||
Date: aws.Time(pastDate),
|
||||
},
|
||||
}},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const oldKey = "d/in-prefix.txt"
|
||||
const otherKey = "other/skip.txt"
|
||||
putObject(t, c, bucket, oldKey, "old")
|
||||
putObject(t, c, bucket, otherKey, "other")
|
||||
|
||||
// Backdate oldKey too — defense-in-depth so the test doesn't
|
||||
// depend on whether the worker also considers MinTriggerAge for
|
||||
// date kinds; a fresh object with a past date should still expire,
|
||||
// but aging it locks the assertion either way.
|
||||
backdateMtime(t, fc, bucket, oldKey, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(oldKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "%s/%s must be expired by past-date rule", bucket, oldKey)
|
||||
|
||||
// Out-of-prefix object stays — the rule's Filter.Prefix gates this.
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(otherKey),
|
||||
})
|
||||
require.NoError(t, err, "object outside the rule's prefix must remain")
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
// AbortIncompleteMultipartUpload integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// putAbortMPULifecycle wires AbortIncompleteMultipartUpload on a prefix.
|
||||
// AWS rejects DaysAfterInitiation < 1, so the test backdates the upload
|
||||
// directory to age it past the threshold.
|
||||
func putAbortMPULifecycle(t *testing.T, c *s3.Client, bucket, prefix string, days int32) {
|
||||
t.Helper()
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{
|
||||
{
|
||||
ID: aws.String("abort-stale-mpu"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String(prefix)},
|
||||
AbortIncompleteMultipartUpload: &types.AbortIncompleteMultipartUpload{
|
||||
DaysAfterInitiation: aws.Int32(days),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// backdateUploadDir ages the .uploads/<uploadID>/ directory entry, which is
|
||||
// what the lifecycle worker keys ABORT_MPU off of (the upload's init-time
|
||||
// directory carries the destination key in Extended).
|
||||
func backdateUploadDir(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket, uploadID string, daysOld int) {
|
||||
t.Helper()
|
||||
dir := bucketsPath + "/" + bucket + "/.uploads"
|
||||
resp, err := fc.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: dir, Name: uploadID,
|
||||
})
|
||||
require.NoError(t, err, "lookup .uploads/%s", uploadID)
|
||||
require.NotNil(t, resp.Entry)
|
||||
require.NotNil(t, resp.Entry.Attributes)
|
||||
require.True(t, resp.Entry.IsDirectory, ".uploads/%s should be a directory", uploadID)
|
||||
|
||||
resp.Entry.Attributes.Mtime = time.Now().Add(-time.Duration(daysOld) * 24 * time.Hour).Unix()
|
||||
resp.Entry.Attributes.MtimeNs = 0
|
||||
_, err = fc.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
|
||||
Directory: dir, Entry: resp.Entry,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// TestLifecycleAbortIncompleteMultipartUpload: an MPU left incomplete past
|
||||
// the AbortIncompleteMultipartUpload.DaysAfterInitiation threshold must be
|
||||
// aborted by the lifecycle worker. Exercises the lifecycleAbortMPU handler
|
||||
// path which is otherwise unreachable from the prefix-based expiration
|
||||
// tests (the routing keys off of `.uploads/<id>/` directory events, not
|
||||
// regular object events).
|
||||
func TestLifecycleAbortIncompleteMultipartUpload(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("abort-mpu")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putAbortMPULifecycle(t, c, bucket, "uploads/", 1)
|
||||
|
||||
// Initiate an MPU, upload a single part, then leave it. CompleteMultipart
|
||||
// would consume the upload; AbortMultipart would explicitly abort. We do
|
||||
// neither so the lifecycle worker is the path that ends it.
|
||||
const key = "uploads/big.bin"
|
||||
createOut, err := c.CreateMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
uploadID := aws.ToString(createOut.UploadId)
|
||||
require.NotEmpty(t, uploadID)
|
||||
|
||||
// Upload one part so the directory carries the right shape.
|
||||
_, err = c.UploadPart(context.Background(), &s3.UploadPartInput{
|
||||
Bucket: aws.String(bucket),
|
||||
Key: aws.String(key),
|
||||
UploadId: aws.String(uploadID),
|
||||
PartNumber: aws.Int32(1),
|
||||
Body: strings.NewReader("part 1 contents"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Sanity: ListMultipartUploads sees our pending upload before the
|
||||
// worker runs.
|
||||
listOut, err := c.ListMultipartUploads(context.Background(), &s3.ListMultipartUploadsInput{
|
||||
Bucket: aws.String(bucket),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
var foundBefore bool
|
||||
for _, u := range listOut.Uploads {
|
||||
if aws.ToString(u.UploadId) == uploadID {
|
||||
foundBefore = true
|
||||
break
|
||||
}
|
||||
}
|
||||
require.True(t, foundBefore, "upload %s should be visible before the worker runs", uploadID)
|
||||
|
||||
// Backdate the upload directory entry so the rule's DaysAfterInitiation
|
||||
// threshold is satisfied. The router walks .uploads/ and emits an
|
||||
// MPU-init event for each entry.
|
||||
backdateUploadDir(t, fc, bucket, uploadID, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// The upload must no longer appear in ListMultipartUploads.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
listOut, err := c.ListMultipartUploads(context.Background(), &s3.ListMultipartUploadsInput{
|
||||
Bucket: aws.String(bucket),
|
||||
})
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, u := range listOut.Uploads {
|
||||
if aws.ToString(u.UploadId) == uploadID {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}, 30*time.Second, 500*time.Millisecond, "upload %s must be aborted by the lifecycle worker", uploadID)
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
// Multiple-bucket integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
smithyhttp "github.com/aws/smithy-go/transport/http"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// isS3NotFound recognizes a NotFound (NoSuchKey/404) response from the
|
||||
// AWS SDK. Treating any HeadObject error as "deleted" lets a transport
|
||||
// failure or dead endpoint mask a real bug, so callers that need to
|
||||
// prove deletion specifically should use this.
|
||||
func isS3NotFound(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
var nsk *types.NoSuchKey
|
||||
if errors.As(err, &nsk) {
|
||||
return true
|
||||
}
|
||||
var notFound *types.NotFound
|
||||
if errors.As(err, ¬Found) {
|
||||
return true
|
||||
}
|
||||
var apiErr *smithyhttp.ResponseError
|
||||
if errors.As(err, &apiErr) {
|
||||
return apiErr.HTTPStatusCode() == 404
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// TestLifecycleMultipleBucketsInOneSweep: a single shell-driven shard
|
||||
// sweep must process every bucket carrying lifecycle config, not just
|
||||
// the first one alphabetically. Pinned because the scheduler iterates
|
||||
// the buckets directory and a regression that returns early after the
|
||||
// first match would silently disable lifecycle for every later bucket.
|
||||
//
|
||||
// Two buckets, each with its own 1-day prefix-expiration rule and one
|
||||
// backdated object. After the worker runs, both objects must be gone.
|
||||
func TestLifecycleMultipleBucketsInOneSweep(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucketA := uniqueBucket("multi-a")
|
||||
bucketB := uniqueBucket("multi-b")
|
||||
mustCreateBucket(t, c, bucketA)
|
||||
mustCreateBucket(t, c, bucketB)
|
||||
|
||||
putExpirationLifecycle(t, c, bucketA, "exp/", 1)
|
||||
putExpirationLifecycle(t, c, bucketB, "exp/", 1)
|
||||
|
||||
const keyA = "exp/a.txt"
|
||||
const keyB = "exp/b.txt"
|
||||
putObject(t, c, bucketA, keyA, "a")
|
||||
putObject(t, c, bucketB, keyB, "b")
|
||||
backdateMtime(t, fc, bucketA, keyA, 30)
|
||||
backdateMtime(t, fc, bucketB, keyB, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// Both buckets must have their objects expired in this single sweep.
|
||||
for _, c2 := range []struct {
|
||||
bucket, key string
|
||||
}{
|
||||
{bucketA, keyA},
|
||||
{bucketB, keyB},
|
||||
} {
|
||||
c2 := c2
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(c2.bucket), Key: aws.String(c2.key),
|
||||
})
|
||||
// Pin: only count 404 / NoSuchKey as deletion. Any other
|
||||
// error (transport failure, dead endpoint, auth) would
|
||||
// otherwise mask a real bug as a "passed" test.
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond,
|
||||
"%s/%s must be expired by the multi-bucket sweep", c2.bucket, c2.key)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
// Multiple-rules-per-bucket integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleMultipleRulesInOneBucket: a single bucket carries two
|
||||
// independent Expiration rules with disjoint prefix filters and
|
||||
// different Days thresholds. Each rule must fire only on its prefix
|
||||
// and ignore the other; objects outside both prefixes must survive.
|
||||
//
|
||||
// Pinned because compile.Compile builds one CompiledAction per rule
|
||||
// per kind, all sharing the same bucket index. A bug that lets one
|
||||
// rule's prefix or threshold leak into the other (e.g. last-write-
|
||||
// wins on a shared map) would silently expire wrong objects.
|
||||
func TestLifecycleMultipleRulesInOneBucket(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("multi-rule")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{
|
||||
{
|
||||
ID: aws.String("logs-1d"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String("logs/")},
|
||||
Expiration: &types.LifecycleExpiration{Days: aws.Int32(1)},
|
||||
},
|
||||
{
|
||||
ID: aws.String("tmp-7d"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String("tmp/")},
|
||||
Expiration: &types.LifecycleExpiration{Days: aws.Int32(7)},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const logsKey = "logs/access.log"
|
||||
const tmpKey = "tmp/scratch.bin"
|
||||
const otherKey = "data/keep.bin"
|
||||
|
||||
putObject(t, c, bucket, logsKey, "log entry")
|
||||
putObject(t, c, bucket, tmpKey, "scratch")
|
||||
putObject(t, c, bucket, otherKey, "important")
|
||||
|
||||
// Backdate logs/ past the 1d rule, tmp/ past the 7d rule, data/
|
||||
// past both thresholds — but data/ matches no rule, so it must
|
||||
// survive regardless of age.
|
||||
backdateMtime(t, fc, bucket, logsKey, 30)
|
||||
backdateMtime(t, fc, bucket, tmpKey, 30)
|
||||
backdateMtime(t, fc, bucket, otherKey, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// logs/ rule fires.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(logsKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "logs/ rule must expire %s", logsKey)
|
||||
|
||||
// tmp/ rule fires.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(tmpKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "tmp/ rule must expire %s", tmpKey)
|
||||
|
||||
// data/ matches NEITHER rule and must survive — pinning that the
|
||||
// per-rule prefix gate is independent and additive, not a global
|
||||
// "any rule's prefix" check.
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(otherKey),
|
||||
})
|
||||
require.NoError(t, err, "object outside both rule prefixes must survive")
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
// NewerNoncurrentVersions integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// putNewerNoncurrentLifecycle keeps the N most recent noncurrent versions
|
||||
// and expires the rest. NoncurrentDays is required by the AWS XML schema
|
||||
// alongside NewerNoncurrentVersions; keeping it at 1 means the rule fires
|
||||
// as soon as a version moves past the keep-count and is older than 1 day.
|
||||
func putNewerNoncurrentLifecycle(t *testing.T, c *s3.Client, bucket, prefix string, keep int32) {
|
||||
t.Helper()
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{
|
||||
{
|
||||
ID: aws.String("keep-newest-noncurrent"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String(prefix)},
|
||||
NoncurrentVersionExpiration: &types.NoncurrentVersionExpiration{
|
||||
NewerNoncurrentVersions: aws.Int32(keep),
|
||||
NoncurrentDays: aws.Int32(1),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// TestLifecycleNewerNoncurrentVersions: NewerNoncurrentVersions=1 keeps
|
||||
// only the most recent noncurrent version; older noncurrent versions
|
||||
// must be deleted. PUT v1, v2, v3, v4 (v4 current; v1-v3 noncurrent),
|
||||
// backdate v1, v2, v3, run the worker, expect v3 (newest noncurrent)
|
||||
// and v4 (current) to remain, v1 and v2 to be removed.
|
||||
//
|
||||
// This routes through routePointerTransition's "needs full expansion"
|
||||
// path — distinct from the per-version NoncurrentDays path because
|
||||
// the rule depends on per-version rank, not just per-version age.
|
||||
func TestLifecycleNewerNoncurrentVersions(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("newer-nc")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putVersioningEnabled(t, c, bucket)
|
||||
putNewerNoncurrentLifecycle(t, c, bucket, "n/", 1)
|
||||
|
||||
const key = "n/obj.txt"
|
||||
puts := make([]string, 4) // versionIds in PUT order
|
||||
for i := range puts {
|
||||
out, err := c.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
Body: strings.NewReader(time.Now().Format(time.RFC3339Nano) + "-" + string(rune('a'+i))),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
puts[i] = aws.ToString(out.VersionId)
|
||||
require.NotEmpty(t, puts[i])
|
||||
}
|
||||
v1, v2, v3, v4 := puts[0], puts[1], puts[2], puts[3]
|
||||
require.NotEqual(t, v1, v2)
|
||||
require.NotEqual(t, v3, v4)
|
||||
|
||||
// Backdate every noncurrent (v1, v2, v3) so all three satisfy the
|
||||
// NoncurrentDays>=1 floor; the keep-newest-1 logic then chooses
|
||||
// which to drop.
|
||||
for _, vid := range []string{v1, v2, v3} {
|
||||
backdateVersionedMtime(t, fc, bucket, key, vid, 30)
|
||||
}
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// v1 and v2 (older noncurrent) must be expired.
|
||||
for _, vid := range []string{v1, v2} {
|
||||
vid := vid
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(vid),
|
||||
})
|
||||
return err != nil
|
||||
}, 30*time.Second, 500*time.Millisecond, "older noncurrent version %s must be expired", vid)
|
||||
}
|
||||
|
||||
// v3 (newest noncurrent within the keep=1 window) must remain.
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(v3),
|
||||
})
|
||||
require.NoError(t, err, "v3 (newest noncurrent within keep=1) must survive")
|
||||
|
||||
// v4 (current) must remain — current versions are immune to
|
||||
// NoncurrentVersionExpiration regardless of how the keep-count
|
||||
// arithmetic plays out.
|
||||
currentHead, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err, "current version v4 must remain")
|
||||
require.Equal(t, v4, aws.ToString(currentHead.VersionId))
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
// Object Lock + lifecycle integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleSkipsObjectLockedObjects: an object under Object Lock
|
||||
// retention must NOT be expired by the lifecycle worker. The handler's
|
||||
// enforceObjectLockProtections check (s3api_internal_lifecycle.go:47)
|
||||
// returns an error which the dispatcher classifies as
|
||||
// SKIPPED_OBJECT_LOCK. Pinning the integration end-to-end so a
|
||||
// regression in either the lock state lookup or the dispatcher's
|
||||
// outcome handling is caught.
|
||||
//
|
||||
// Companion positive control: a second object in the same bucket
|
||||
// without retention is expired by the same sweep — proves the rule
|
||||
// is firing AND that the lock check is selective, not a blanket skip.
|
||||
func TestLifecycleSkipsObjectLockedObjects(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("objlock")
|
||||
// Object Lock requires the bucket to be created with the flag set;
|
||||
// it can't be enabled retroactively on an existing bucket. Use the
|
||||
// raw API rather than mustCreateBucket here so we can pass the
|
||||
// ObjectLockEnabledForBucket option.
|
||||
_, err := c.CreateBucket(context.Background(), &s3.CreateBucketInput{
|
||||
Bucket: aws.String(bucket),
|
||||
ObjectLockEnabledForBucket: aws.Bool(true),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
// Best-effort cleanup mirrors mustCreateBucket; locked objects
|
||||
// may resist deletion but the test process continues regardless.
|
||||
c.DeleteBucketLifecycle(context.Background(), &s3.DeleteBucketLifecycleInput{Bucket: aws.String(bucket)})
|
||||
listOut, _ := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{Bucket: aws.String(bucket)})
|
||||
if listOut != nil {
|
||||
for _, v := range listOut.Versions {
|
||||
// Bypass governance retention to free the bucket for delete.
|
||||
c.DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(bucket),
|
||||
Key: v.Key,
|
||||
VersionId: v.VersionId,
|
||||
BypassGovernanceRetention: aws.Bool(true),
|
||||
})
|
||||
}
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
c.DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(bucket), Key: m.Key, VersionId: m.VersionId,
|
||||
})
|
||||
}
|
||||
}
|
||||
c.DeleteBucket(context.Background(), &s3.DeleteBucketInput{Bucket: aws.String(bucket)})
|
||||
})
|
||||
|
||||
putExpirationLifecycle(t, c, bucket, "lock/", 1)
|
||||
|
||||
const lockedKey = "lock/protected.txt"
|
||||
const freeKey = "lock/free.txt"
|
||||
|
||||
// Locked object: PUT with GOVERNANCE retention until 1h from now.
|
||||
lockedPut, err := c.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket),
|
||||
Key: aws.String(lockedKey),
|
||||
Body: strings.NewReader("locked"),
|
||||
ObjectLockMode: types.ObjectLockModeGovernance,
|
||||
ObjectLockRetainUntilDate: aws.Time(time.Now().Add(time.Hour)),
|
||||
})
|
||||
require.NoError(t, err, "PUT with retention must succeed on a lock-enabled bucket")
|
||||
require.NotEmpty(t, aws.ToString(lockedPut.VersionId))
|
||||
|
||||
// Free object: PUT without retention. Object Lock requires
|
||||
// versioning, so the bucket is implicitly versioned and every PUT
|
||||
// produces a versionId. Capture both for the version-aware
|
||||
// backdate path.
|
||||
freePut, err := c.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(freeKey), Body: strings.NewReader("free"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
freeVersionID := aws.ToString(freePut.VersionId)
|
||||
require.NotEmpty(t, freeVersionID)
|
||||
|
||||
// Backdate both so they would otherwise both expire under the
|
||||
// 1-day rule. The lock check is what distinguishes them.
|
||||
// Versioning-enabled buckets store entries under .versions/v_<id>,
|
||||
// not at the bare key path, so use backdateVersionedMtime.
|
||||
backdateVersionedMtime(t, fc, bucket, lockedKey, aws.ToString(lockedPut.VersionId), 30)
|
||||
backdateVersionedMtime(t, fc, bucket, freeKey, freeVersionID, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// Free object expires (positive control).
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(freeKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "unlocked object must expire (control)")
|
||||
|
||||
// Locked object survives — Object Lock retention overrides the
|
||||
// lifecycle rule. HEAD without versionId returns the still-current
|
||||
// object data, not a 404.
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(lockedKey),
|
||||
})
|
||||
require.NoError(t, err, "object under GOVERNANCE retention must survive the sweep")
|
||||
}
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
// Size-filter integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestLifecycleSizeFilterGreaterThan: ObjectSizeGreaterThan is a strict >
|
||||
// gate (filterAllows in match.go uses ev.Size <= rule.FilterSizeGreaterThan
|
||||
// to reject). Two backdated objects on the same prefix — only the one
|
||||
// strictly larger than the threshold is expired.
|
||||
func TestLifecycleSizeFilterGreaterThan(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("size-filter")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
|
||||
const threshold = 100
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{{
|
||||
ID: aws.String("size-gt"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{
|
||||
And: &types.LifecycleRuleAndOperator{
|
||||
Prefix: aws.String("sz/"),
|
||||
ObjectSizeGreaterThan: aws.Int64(threshold),
|
||||
},
|
||||
},
|
||||
Expiration: &types.LifecycleExpiration{Days: aws.Int32(1)},
|
||||
}},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const smallKey = "sz/small.txt" // size <= threshold → must remain
|
||||
const largeKey = "sz/large.txt" // size > threshold → must expire
|
||||
putObject(t, c, bucket, smallKey, strings.Repeat("a", threshold)) // exactly at boundary; gate is strictly >, so rejects
|
||||
putObject(t, c, bucket, largeKey, strings.Repeat("a", threshold+50))
|
||||
|
||||
backdateMtime(t, fc, bucket, smallKey, 30)
|
||||
backdateMtime(t, fc, bucket, largeKey, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// Larger-than-threshold object expired.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(largeKey),
|
||||
})
|
||||
return err != nil
|
||||
}, 30*time.Second, 500*time.Millisecond, "object larger than threshold must expire")
|
||||
|
||||
// Boundary object (size == threshold) survives — strictly-greater gate.
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(smallKey),
|
||||
})
|
||||
require.NoError(t, err, "object at exact threshold must remain (gate is strictly >)")
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
// Suspended-versioning Expiration integration scenario.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// putVersioningSuspended flips Status=Suspended. PUTs after this point
|
||||
// overwrite the null version in place rather than creating new versionIds.
|
||||
func putVersioningSuspended(t *testing.T, c *s3.Client, bucket string) {
|
||||
t.Helper()
|
||||
_, err := c.PutBucketVersioning(context.Background(), &s3.PutBucketVersioningInput{
|
||||
Bucket: aws.String(bucket),
|
||||
VersioningConfiguration: &types.VersioningConfiguration{Status: types.BucketVersioningStatusSuspended},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// TestLifecycleSuspendedVersioningExpiration: on a suspended-versioning
|
||||
// bucket, Expiration takes a different code path than the Enabled and
|
||||
// Off cases. lifecycleDispatch's VersioningSuspended branch first
|
||||
// deletes the null version (via deleteSpecificObjectVersion(versionId="null"))
|
||||
// and then writes a fresh delete marker on top.
|
||||
//
|
||||
// Setup: enable then suspend versioning so any noncurrent versions can
|
||||
// exist alongside the null one; PUT once-with-versioning + once-after-
|
||||
// suspend (so we have a noncurrent version + a null version), backdate
|
||||
// the null, run the worker. The null must be deleted and replaced by a
|
||||
// delete marker; the prior noncurrent version stays addressable.
|
||||
func TestLifecycleSuspendedVersioningExpiration(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("suspended")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putVersioningEnabled(t, c, bucket)
|
||||
|
||||
const key = "s/obj.txt"
|
||||
// First PUT under Enabled: produces a real versionId.
|
||||
putObject(t, c, bucket, key, "v1")
|
||||
headEnabled, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
v1 := aws.ToString(headEnabled.VersionId)
|
||||
require.NotEmpty(t, v1)
|
||||
require.NotEqual(t, "null", v1)
|
||||
|
||||
// Suspend versioning, then PUT again. The new write becomes the
|
||||
// "null" version that overwrites the bare-key entry; v1 demotes to
|
||||
// a noncurrent version.
|
||||
putVersioningSuspended(t, c, bucket)
|
||||
putObject(t, c, bucket, key, "null-version")
|
||||
|
||||
// Set the Expiration rule AFTER the writes so the rule
|
||||
// configuration race against PUTs is deterministic. Days=1 for
|
||||
// AWS-spec compliance; backdating the null entry makes the worker
|
||||
// fire immediately.
|
||||
putExpirationLifecycle(t, c, bucket, "s/", 1)
|
||||
|
||||
// Backdate the bare-key entry (the null version): on suspended
|
||||
// buckets the null lives at the bare path, NOT under .versions/.
|
||||
backdateMtime(t, fc, bucket, key, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// HEAD without versionId hits the latest, which after the worker
|
||||
// runs should be a delete marker (404 NoSuchKey).
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
return err != nil
|
||||
}, 30*time.Second, 500*time.Millisecond, "delete marker must dominate latest after suspended-versioning expiration")
|
||||
|
||||
// The noncurrent v1 (created under Enabled) must remain — Expiration
|
||||
// in the suspended branch only touches the null version, not other
|
||||
// versions in the .versions/ folder.
|
||||
directHead, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(v1),
|
||||
})
|
||||
require.NoError(t, err, "noncurrent v1 must survive — suspended Expiration only touches the null")
|
||||
require.NotNil(t, directHead)
|
||||
|
||||
// ListObjectVersions must show: v1 still present, a fresh delete
|
||||
// marker as latest, and no more "null" version.
|
||||
listOut, err := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{
|
||||
Bucket: aws.String(bucket), Prefix: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
var sawV1, sawMarker bool
|
||||
for _, v := range listOut.Versions {
|
||||
if aws.ToString(v.Key) == key {
|
||||
vid := aws.ToString(v.VersionId)
|
||||
if vid == v1 {
|
||||
sawV1 = true
|
||||
}
|
||||
require.NotEqual(t, "null", vid, "null version must have been removed")
|
||||
}
|
||||
}
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
if aws.ToString(m.Key) == key && aws.ToBool(m.IsLatest) {
|
||||
sawMarker = true
|
||||
}
|
||||
}
|
||||
require.True(t, sawV1, "v1 must still appear in ListObjectVersions")
|
||||
require.True(t, sawMarker, "fresh delete marker must dominate latest")
|
||||
}
|
||||
@@ -90,13 +90,50 @@ func mustCreateBucket(t *testing.T, c *s3.Client, name string) {
|
||||
_, err := c.CreateBucket(context.Background(), &s3.CreateBucketInput{Bucket: aws.String(name)})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
// Best effort: empty + delete.
|
||||
listOut, _ := c.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{Bucket: aws.String(name)})
|
||||
// Drop the lifecycle configuration first so any subsequent
|
||||
// shell-driven shard sweep stops loading rules for this bucket
|
||||
// — without this, a later test's run-shard would pick up the
|
||||
// dead bucket's config and produce phantom dispatches.
|
||||
c.DeleteBucketLifecycle(context.Background(), &s3.DeleteBucketLifecycleInput{Bucket: aws.String(name)})
|
||||
|
||||
// Empty every version + delete marker (versioning-aware buckets
|
||||
// hold state that ListObjectsV2 doesn't surface). Best-effort:
|
||||
// errors are tolerated because the bucket might already be
|
||||
// half-torn-down.
|
||||
listOut, _ := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{Bucket: aws.String(name)})
|
||||
if listOut != nil {
|
||||
for _, o := range listOut.Contents {
|
||||
for _, v := range listOut.Versions {
|
||||
c.DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(name), Key: v.Key, VersionId: v.VersionId,
|
||||
})
|
||||
}
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
c.DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(name), Key: m.Key, VersionId: m.VersionId,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Catch any non-versioned objects ListObjectVersions missed.
|
||||
objs, _ := c.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{Bucket: aws.String(name)})
|
||||
if objs != nil {
|
||||
for _, o := range objs.Contents {
|
||||
c.DeleteObject(context.Background(), &s3.DeleteObjectInput{Bucket: aws.String(name), Key: o.Key})
|
||||
}
|
||||
}
|
||||
|
||||
// Abort any in-flight multipart uploads — the AbortMPU lifecycle
|
||||
// test leaves these intentionally; without an explicit Abort the
|
||||
// bucket DELETE refuses with NotEmpty.
|
||||
mpus, _ := c.ListMultipartUploads(context.Background(), &s3.ListMultipartUploadsInput{Bucket: aws.String(name)})
|
||||
if mpus != nil {
|
||||
for _, u := range mpus.Uploads {
|
||||
c.AbortMultipartUpload(context.Background(), &s3.AbortMultipartUploadInput{
|
||||
Bucket: aws.String(name), Key: u.Key, UploadId: u.UploadId,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
c.DeleteBucket(context.Background(), &s3.DeleteBucketInput{Bucket: aws.String(name)})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,441 @@
|
||||
// Versioning + delete-marker + noncurrent integration scenarios for the
|
||||
// event-driven lifecycle worker. Same backdating trick as the base test:
|
||||
// AWS rejects Days < 1, so the test ages the object via the filer's
|
||||
// UpdateEntry RPC and runs the shell command once with -runtime 10s.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// putVersioningEnabled flips Status=Enabled on the bucket so subsequent
|
||||
// PUTs produce versioned siblings rather than overwriting.
|
||||
func putVersioningEnabled(t *testing.T, c *s3.Client, bucket string) {
|
||||
t.Helper()
|
||||
_, err := c.PutBucketVersioning(context.Background(), &s3.PutBucketVersioningInput{
|
||||
Bucket: aws.String(bucket),
|
||||
VersioningConfiguration: &types.VersioningConfiguration{Status: types.BucketVersioningStatusEnabled},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// putNoncurrentExpirationLifecycle sets a NoncurrentVersionExpiration rule
|
||||
// targeting a prefix. Days=1 because AWS rejects 0; the test backdates
|
||||
// the noncurrent version.
|
||||
func putNoncurrentExpirationLifecycle(t *testing.T, c *s3.Client, bucket, prefix string, days int32) {
|
||||
t.Helper()
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{
|
||||
{
|
||||
ID: aws.String("expire-noncurrent"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String(prefix)},
|
||||
NoncurrentVersionExpiration: &types.NoncurrentVersionExpiration{
|
||||
NoncurrentDays: aws.Int32(days),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// backdateVersionedMtime ages a specific .versions/v_<id> entry. The
|
||||
// version files live under <buckets>/<bucket>/<key>.versions/v_<versionId>.
|
||||
func backdateVersionedMtime(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket, key, versionID string, daysOld int) {
|
||||
t.Helper()
|
||||
dir := bucketsPath + "/" + bucket + "/" + key + ".versions"
|
||||
name := "v_" + versionID
|
||||
resp, err := fc.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: dir, Name: name,
|
||||
})
|
||||
require.NoError(t, err, "lookup version %s/%s", dir, name)
|
||||
require.NotNil(t, resp.Entry)
|
||||
require.NotNil(t, resp.Entry.Attributes)
|
||||
|
||||
resp.Entry.Attributes.Mtime = time.Now().Add(-time.Duration(daysOld) * 24 * time.Hour).Unix()
|
||||
resp.Entry.Attributes.MtimeNs = 0
|
||||
_, err = fc.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
|
||||
Directory: dir, Entry: resp.Entry,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func runLifecycleShard(t *testing.T) string {
|
||||
t.Helper()
|
||||
out := runShellCommand(t, fmt.Sprintf(
|
||||
"s3.lifecycle.run-shard -shards 0-15 -s3 %s -events 0 -dispatch 200ms -checkpoint 5s -runtime 10s",
|
||||
envOr("S3_GRPC_ENDPOINT", defaultS3GrpcEndpoint),
|
||||
))
|
||||
require.NotContains(t, out, "FATAL", "shell output:\n%s", out)
|
||||
return out
|
||||
}
|
||||
|
||||
// TestLifecycleVersionedBucketCreatesDeleteMarker: an Expiration rule on a
|
||||
// versioned bucket must produce a delete marker (latest version after
|
||||
// expiration) rather than removing the version itself. Pre-fix this would
|
||||
// have unconditionally deleted the live version.
|
||||
func TestLifecycleVersionedBucketCreatesDeleteMarker(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("vexp")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putVersioningEnabled(t, c, bucket)
|
||||
putExpirationLifecycle(t, c, bucket, "expire/", 1)
|
||||
|
||||
const key = "expire/old.txt"
|
||||
putObject(t, c, bucket, key, "v1")
|
||||
|
||||
// Look up the version we just created so we can backdate it. The bare
|
||||
// filer entry under <bucket>/<key> is the latest-version pointer in
|
||||
// SeaweedFS's versioned layout.
|
||||
headOut, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, headOut.VersionId)
|
||||
versionID := aws.ToString(headOut.VersionId)
|
||||
require.NotEmpty(t, versionID)
|
||||
backdateVersionedMtime(t, fc, bucket, key, versionID, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// HEAD without versionId returns the latest version. After the worker
|
||||
// runs, the latest must be a delete marker (NoSuchKey on plain HEAD).
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "expected delete marker to become latest for %s/%s", bucket, key)
|
||||
|
||||
// The original version must still be addressable directly — Expiration
|
||||
// on a versioned bucket creates a marker, it doesn't drop the version.
|
||||
directHead, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID),
|
||||
})
|
||||
require.NoError(t, err, "original version must still exist")
|
||||
require.NotNil(t, directHead)
|
||||
|
||||
// ListObjectVersions should now show a delete marker dominating the
|
||||
// version. The IsLatest=true marker proves the worker created a fresh
|
||||
// marker rather than aging the existing version.
|
||||
listOut, err := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{
|
||||
Bucket: aws.String(bucket), Prefix: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
var sawMarker bool
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
if aws.ToString(m.Key) == key && aws.ToBool(m.IsLatest) {
|
||||
sawMarker = true
|
||||
break
|
||||
}
|
||||
}
|
||||
require.True(t, sawMarker, "ListObjectVersions must show a delete marker as latest for %s", key)
|
||||
}
|
||||
|
||||
// TestLifecycleNoncurrentVersionExpiration: NoncurrentVersionExpiration
|
||||
// only fires on noncurrent versions. PUT v1, PUT v2 (so v1 → noncurrent),
|
||||
// backdate v1, run worker. v1 must be removed; v2 stays current.
|
||||
func TestLifecycleNoncurrentVersionExpiration(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("noncurrent")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putVersioningEnabled(t, c, bucket)
|
||||
putNoncurrentExpirationLifecycle(t, c, bucket, "v/", 1)
|
||||
|
||||
const key = "v/obj.txt"
|
||||
// First PUT.
|
||||
put1, err := c.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), Body: strings.NewReader("v1"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
v1 := aws.ToString(put1.VersionId)
|
||||
|
||||
// Second PUT promotes v1 to noncurrent.
|
||||
put2, err := c.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), Body: strings.NewReader("v2"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
v2 := aws.ToString(put2.VersionId)
|
||||
require.NotEqual(t, v1, v2)
|
||||
|
||||
// NoncurrentDays is clocked from the SUCCESSOR's mtime — i.e. when
|
||||
// the version became noncurrent — not from the displaced version's
|
||||
// own mtime. Backdating only v1 leaves the noncurrent clock (v2's
|
||||
// mtime) at "now" and the rule never fires. Age both with v1 older
|
||||
// than v2 so the ordering is realistic.
|
||||
backdateVersionedMtime(t, fc, bucket, key, v1, 31)
|
||||
backdateVersionedMtime(t, fc, bucket, key, v2, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// v1 must be gone.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(v1),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "noncurrent v1 must be expired")
|
||||
|
||||
// v2 must still be addressable BY VERSION ID — pinning that the
|
||||
// noncurrent dispatch didn't accidentally remove the current
|
||||
// version's file.
|
||||
directHead, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(v2),
|
||||
})
|
||||
require.NoError(t, err, "current version v2 (versionId=%s) must still be addressable directly", v2)
|
||||
|
||||
// And HEAD without versionId (i.e., the latest) must hit v2 — pinning
|
||||
// that the bare-key pointer wasn't removed and a delete marker
|
||||
// wasn't fabricated as a side effect of the noncurrent delete. On
|
||||
// failure we dump ListObjectVersions so the failure log shows
|
||||
// exactly what's there (versions vs delete markers vs bare).
|
||||
currentHead, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
if err != nil {
|
||||
listOut, listErr := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{
|
||||
Bucket: aws.String(bucket), Prefix: aws.String(key),
|
||||
})
|
||||
if listErr != nil || listOut == nil {
|
||||
t.Logf("HeadObject(latest) failed: %v\nListObjectVersions err=%v", err, listErr)
|
||||
} else {
|
||||
t.Logf("HeadObject(latest) failed: %v\nListObjectVersions err=%v versions=%d markers=%d", err, listErr, len(listOut.Versions), len(listOut.DeleteMarkers))
|
||||
for _, v := range listOut.Versions {
|
||||
t.Logf(" version key=%s id=%s isLatest=%v", aws.ToString(v.Key), aws.ToString(v.VersionId), aws.ToBool(v.IsLatest))
|
||||
}
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
t.Logf(" marker key=%s id=%s isLatest=%v", aws.ToString(m.Key), aws.ToString(m.VersionId), aws.ToBool(m.IsLatest))
|
||||
}
|
||||
}
|
||||
require.NoError(t, err, "HEAD(latest) on %s/%s must succeed; v2 (%s) must remain current", bucket, key, v2)
|
||||
}
|
||||
require.Equal(t, v2, aws.ToString(currentHead.VersionId), "latest pointer must still resolve to v2")
|
||||
_ = directHead
|
||||
}
|
||||
|
||||
// TestLifecycleExpiredDeleteMarkerCleanup: a sole-survivor delete marker
|
||||
// must be removed by an ExpiredObjectDeleteMarker=true rule. Setup: PUT v1,
|
||||
// DELETE (creates a marker that becomes latest, with v1 as noncurrent),
|
||||
// expire v1 via NoncurrentVersionExpiration so the marker is alone, then
|
||||
// the marker rule fires.
|
||||
//
|
||||
// We exercise this in two passes: pass 1 (Noncurrent rule on a sibling
|
||||
// prefix) cleans v1; pass 2 (ExpiredObjectDeleteMarker rule, also on the
|
||||
// same prefix) removes the now-orphaned marker. Since the lifecycle XML
|
||||
// only carries one set of rules per bucket, we wire both.
|
||||
func TestLifecycleExpiredDeleteMarkerCleanup(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("marker")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
putVersioningEnabled(t, c, bucket)
|
||||
|
||||
// Combined rule: noncurrent expiration + delete-marker cleanup, both
|
||||
// on the same prefix.
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{
|
||||
{
|
||||
ID: aws.String("expire-noncurrent"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String("m/")},
|
||||
NoncurrentVersionExpiration: &types.NoncurrentVersionExpiration{
|
||||
NoncurrentDays: aws.Int32(1),
|
||||
},
|
||||
},
|
||||
{
|
||||
ID: aws.String("expire-marker"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String("m/")},
|
||||
Expiration: &types.LifecycleExpiration{
|
||||
ExpiredObjectDeleteMarker: aws.Bool(true),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const key = "m/obj.txt"
|
||||
put1, err := c.PutObject(context.Background(), &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key), Body: strings.NewReader("v1"),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
v1 := aws.ToString(put1.VersionId)
|
||||
|
||||
// Plain DELETE with versioning on creates a marker; the marker becomes
|
||||
// latest, v1 demotes to noncurrent.
|
||||
_, err = c.DeleteObject(context.Background(), &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Backdate v1 so the noncurrent rule expires it.
|
||||
backdateVersionedMtime(t, fc, bucket, key, v1, 30)
|
||||
|
||||
// Backdate the delete marker too. ListObjectVersions tells us the
|
||||
// marker's versionId.
|
||||
listOut, err := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{
|
||||
Bucket: aws.String(bucket), Prefix: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
var markerID string
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
if aws.ToString(m.Key) == key {
|
||||
markerID = aws.ToString(m.VersionId)
|
||||
break
|
||||
}
|
||||
}
|
||||
require.NotEmpty(t, markerID, "expected to find the delete marker we just created")
|
||||
backdateVersionedMtime(t, fc, bucket, key, markerID, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// Both v1 and the marker must be gone — the .versions/<key>/ folder
|
||||
// should hold nothing for this key.
|
||||
require.Eventuallyf(t, func() bool {
|
||||
listOut, err := c.ListObjectVersions(context.Background(), &s3.ListObjectVersionsInput{
|
||||
Bucket: aws.String(bucket), Prefix: aws.String(key),
|
||||
})
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, v := range listOut.Versions {
|
||||
if aws.ToString(v.Key) == key {
|
||||
return false
|
||||
}
|
||||
}
|
||||
for _, m := range listOut.DeleteMarkers {
|
||||
if aws.ToString(m.Key) == key {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}, 30*time.Second, 500*time.Millisecond, "every version and marker for %s must be gone", key)
|
||||
}
|
||||
|
||||
// TestLifecycleDisabledRuleSkipsObject: a rule with Status=Disabled must
|
||||
// not produce dispatches, even on a backdated object that would otherwise
|
||||
// match. Negative test for the engine's enabled-status gate.
|
||||
func TestLifecycleDisabledRuleSkipsObject(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("disabled")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{{
|
||||
ID: aws.String("disabled-rule"),
|
||||
Status: types.ExpirationStatusDisabled,
|
||||
Filter: &types.LifecycleRuleFilter{Prefix: aws.String("d/")},
|
||||
Expiration: &types.LifecycleExpiration{Days: aws.Int32(1)},
|
||||
}},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const key = "d/obj.txt"
|
||||
putObject(t, c, bucket, key, "still-here")
|
||||
backdateMtime(t, fc, bucket, key, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
// Wait the same window; the object MUST still exist after.
|
||||
time.Sleep(2 * time.Second) // give the worker a chance to (incorrectly) act
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(key),
|
||||
})
|
||||
require.NoError(t, err, "Disabled rule must not delete the object")
|
||||
}
|
||||
|
||||
// TestLifecycleTagFilter: a rule with a tag filter must only match objects
|
||||
// carrying that tag. Two backdated objects, one tagged, one not — only the
|
||||
// tagged one is removed.
|
||||
func TestLifecycleTagFilter(t *testing.T) {
|
||||
c := s3Client(t)
|
||||
fc, fcClose := filerClient(t)
|
||||
defer fcClose()
|
||||
|
||||
bucket := uniqueBucket("tagfilter")
|
||||
mustCreateBucket(t, c, bucket)
|
||||
_, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{
|
||||
Bucket: aws.String(bucket),
|
||||
LifecycleConfiguration: &types.BucketLifecycleConfiguration{
|
||||
Rules: []types.LifecycleRule{{
|
||||
ID: aws.String("tag-only"),
|
||||
Status: types.ExpirationStatusEnabled,
|
||||
Filter: &types.LifecycleRuleFilter{
|
||||
And: &types.LifecycleRuleAndOperator{
|
||||
Prefix: aws.String("t/"),
|
||||
Tags: []types.Tag{
|
||||
{Key: aws.String("env"), Value: aws.String("temp")},
|
||||
},
|
||||
},
|
||||
},
|
||||
Expiration: &types.LifecycleExpiration{Days: aws.Int32(1)},
|
||||
}},
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const taggedKey = "t/tagged.txt"
|
||||
const untaggedKey = "t/untagged.txt"
|
||||
putObject(t, c, bucket, taggedKey, "tagged")
|
||||
putObject(t, c, bucket, untaggedKey, "untagged")
|
||||
|
||||
// Stamp the tag on one of them.
|
||||
_, err = c.PutObjectTagging(context.Background(), &s3.PutObjectTaggingInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(taggedKey),
|
||||
Tagging: &types.Tagging{TagSet: []types.Tag{
|
||||
{Key: aws.String("env"), Value: aws.String("temp")},
|
||||
}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
backdateMtime(t, fc, bucket, taggedKey, 30)
|
||||
backdateMtime(t, fc, bucket, untaggedKey, 30)
|
||||
|
||||
out := runLifecycleShard(t)
|
||||
t.Logf("shell output:\n%s", out)
|
||||
|
||||
require.Eventuallyf(t, func() bool {
|
||||
_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(taggedKey),
|
||||
})
|
||||
return isS3NotFound(err)
|
||||
}, 30*time.Second, 500*time.Millisecond, "tagged object must be expired")
|
||||
|
||||
_, err = c.HeadObject(context.Background(), &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket), Key: aws.String(untaggedKey),
|
||||
})
|
||||
require.NoError(t, err, "untagged object must remain")
|
||||
}
|
||||
@@ -135,10 +135,16 @@ func walkEntry(ctx context.Context, snap *engine.Snapshot, bucket string, entry
|
||||
if action == nil {
|
||||
continue
|
||||
}
|
||||
// SCAN_AT_DATE runs its own date-triggered bootstrap. DISABLED can
|
||||
// be flipped at runtime independent of XML Status, so skip it even
|
||||
// though EvaluateAction would also reject.
|
||||
if action.Mode == engine.ModeScanAtDate || action.Mode == engine.ModeDisabled {
|
||||
// DISABLED can be flipped at runtime independent of XML Status,
|
||||
// so skip it even though EvaluateAction would also reject.
|
||||
// SCAN_AT_DATE actions are processed here too — the date check
|
||||
// in EvaluateAction (now.Before(rule.ExpirationDate)) gates the
|
||||
// dispatch, so pre-date walks are no-ops and post-date walks
|
||||
// expire eligible objects. The earlier "scan-at-date runs its
|
||||
// own bootstrap" plan was never wired; until that lands, the
|
||||
// regular bootstrap walk is the only path that fires
|
||||
// ExpirationDate rules.
|
||||
if action.Mode == engine.ModeDisabled {
|
||||
continue
|
||||
}
|
||||
// (kind, info) shape gate: ABORT_MPU only on MPU init records,
|
||||
|
||||
@@ -164,9 +164,10 @@ func TestWalk_NotYetDueSkipped(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalk_DateActionsSkipped(t *testing.T) {
|
||||
// Date kind is handled by its own SCAN_AT_DATE bootstrap, not by the
|
||||
// regular bootstrap walker.
|
||||
func TestWalk_DateActionFiresAfterDate(t *testing.T) {
|
||||
// The dedicated SCAN_AT_DATE bootstrap was never wired; until it
|
||||
// lands, the regular bootstrap walker is the only path that fires
|
||||
// ExpirationDate rules. Entries past the rule's date must dispatch.
|
||||
date := mustTime(t, "2025-06-15T00:00:00Z")
|
||||
rule := &s3lifecycle.Rule{
|
||||
ID: "d",
|
||||
@@ -181,8 +182,30 @@ func TestWalk_DateActionsSkipped(t *testing.T) {
|
||||
}), rec, WalkOptions{Now: date.AddDate(0, 1, 0)}); err != nil {
|
||||
t.Fatalf("Walk: %v", err)
|
||||
}
|
||||
if len(rec.calls) != 1 || rec.calls[0].path != "x/a" {
|
||||
t.Fatalf("date kind past the rule date must dispatch, got %v", rec.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalk_DateActionSkippedBeforeDate(t *testing.T) {
|
||||
// Pre-date walks are no-ops: EvaluateAction returns ActionNone when
|
||||
// now is before rule.ExpirationDate.
|
||||
date := mustTime(t, "2025-06-15T00:00:00Z")
|
||||
rule := &s3lifecycle.Rule{
|
||||
ID: "d",
|
||||
Status: s3lifecycle.StatusEnabled,
|
||||
ExpirationDate: date,
|
||||
}
|
||||
snap := compileEvDriven(t, "bk", rule)
|
||||
|
||||
rec := &recorder{}
|
||||
if _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{
|
||||
{Path: "x/a", IsLatest: true, ModTime: mustTime(t, "2024-01-01T00:00:00Z")},
|
||||
}), rec, WalkOptions{Now: date.AddDate(0, -1, 0)}); err != nil {
|
||||
t.Fatalf("Walk: %v", err)
|
||||
}
|
||||
if len(rec.calls) != 0 {
|
||||
t.Fatalf("date kind should not dispatch from walker, got %v", rec.calls)
|
||||
t.Fatalf("pre-date walk must not dispatch, got %v", rec.calls)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -64,7 +64,14 @@ func (e *Engine) Compile(inputs []CompileInput, opts CompileOptions) *Snapshot {
|
||||
if !hasPrior || mode == ModeUnspecified {
|
||||
mode = decideMode(rule, kind, opts.MetaLogRetention, opts.BootstrapLookbackMin)
|
||||
}
|
||||
active := prior.BootstrapComplete && mode == ModeEventDriven
|
||||
// Active gates routing: MatchPath / MatchOriginalWrite skip
|
||||
// !IsActive actions. ScanAtDate's only dispatch path is the
|
||||
// bootstrap walk's MatchPath call, so the action must be
|
||||
// considered active there or its rule is silently a no-op.
|
||||
// Bootstrap-completion state is per-action and event-driven-
|
||||
// shaped; date-based actions don't need a bootstrap rendezvous.
|
||||
active := mode == ModeScanAtDate ||
|
||||
(prior.BootstrapComplete && mode == ModeEventDriven)
|
||||
|
||||
ca := &CompiledAction{
|
||||
Rule: rule,
|
||||
|
||||
@@ -145,13 +145,18 @@ func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now tim
|
||||
if !info.IsMPUInit && key.ActionKind == s3lifecycle.ActionKindAbortMPU {
|
||||
continue
|
||||
}
|
||||
// Schedule from ModTime, not the meta-log event time: a backdated
|
||||
// or out-of-band entry update has eventTime ≈ now but ModTime far
|
||||
// in the past, so eventTime+Delay would push the dispatch into the
|
||||
// future even though the rule fires immediately. ModTime+Delay is
|
||||
// the correct fire moment; the dispatcher's identity-CAS catches
|
||||
// drift if the object changes meanwhile.
|
||||
dueTime := info.ModTime.Add(action.Delay)
|
||||
// Schedule from the per-kind due moment. ExpirationDate is
|
||||
// rule-relative (the date IS the moment); other kinds are
|
||||
// ModTime-relative. Using ModTime+Delay for ExpirationDate
|
||||
// (Delay=0) puts dueTime at the entry's mtime — a backdated
|
||||
// object's mtime is BEFORE the rule's date, so the eligibility
|
||||
// check below would skip it. ComputeDueAt encapsulates both
|
||||
// shapes; the dispatcher's identity-CAS catches drift if the
|
||||
// object changes meanwhile.
|
||||
dueTime := s3lifecycle.ComputeDueAt(action.Rule, key.ActionKind, info)
|
||||
if dueTime.IsZero() {
|
||||
continue
|
||||
}
|
||||
res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
|
||||
if res.Action == s3lifecycle.ActionNone {
|
||||
continue
|
||||
|
||||
Reference in New Issue
Block a user