feat(shell): s3.lifecycle.run-shard + integration test (#9361)

* feat(shell): s3.lifecycle.run-shard for manual Phase 3 dispatch

Subscribes to the filer meta-log filtered to one (bucket, key-prefix-hash)
shard, routes events through the compiled lifecycle engine, and dispatches
due actions to the S3 server's LifecycleDelete RPC. Persists the per-shard
cursor to /etc/s3/lifecycle/cursors/shard-NN.json so subsequent runs resume.

Operator-runnable harness for end-to-end Phase 3 validation while the
plugin-worker auto-scheduler is still pending. EventBudget bounds a single
invocation; flags expose dispatch + checkpoint cadence.

Discovers buckets by walking the configured DirBuckets path and reading
each bucket entry's Extended[s3-bucket-lifecycle-configuration-xml]
through lifecycle_xml.ParseCanonical. All compiled actions are seeded
BootstrapComplete=true so the run dispatches whatever fires immediately;
production bootstrap walks set this incrementally per bucket.

* test(s3/lifecycle): integration test driving the run-shard shell command

Spins up 'weed mini', creates a bucket with a 1-day expiration on a prefix,
PUTs the target object, then rewrites the entry's Mtime via filer
UpdateEntry to 30 days ago. Runs 's3.lifecycle.run-shard' for every
shard via 'weed shell' subprocess and asserts the backdated object is
deleted within 30s, and the in-prefix-but-recent object remains.

The S3 API rejects Expiration.Days < 1, so 'wait a day' is unworkable.
Backdating via the filer's gRPC sidesteps that constraint while still
exercising the real Reader -> Router -> Schedule -> Dispatcher ->
LifecycleDelete RPC path end-to-end.

Wires a new s3-lifecycle-tests job into s3-go-tests.yml. The test runs
all 16 shards because ShardID(bucket, key) is hash-based and the test
shouldn't couple to that detail; running every shard keeps the test
independent of the hash function.

* fix(shell/s3.lifecycle.run-shard): address review findings

- Reject negative -events explicitly. Help text already defines 0 as
  unbounded; negative budgets created ambiguous behavior in pipeline.Run.
- Bound the gRPC dial with a 30s timeout instead of context.Background()
  so an unreachable S3 endpoint doesn't hang the shell.
- Paginate the bucket listing in loadLifecycleCompileInputs. SeaweedList
  takes a single-RPC limit; the prior 4096 silently dropped buckets
  past that page on large clusters. Loop with startFrom until a page
  comes back short.
- Surface parse errors instead of swallowing them. Buckets with
  malformed lifecycle XML now print the first three errors verbatim
  and a count for the rest, so an operator running this command for
  diagnostics can find what's wrong.

* feat(shell/s3.lifecycle.run-shard): -shards range/set with one subscription

Adds -shards "lo-hi" or "a,b,c" to the manual run command and threads
the same model through Reader and Pipeline.

- reader.Reader gains ShardPredicate (func(int) bool) and StartTsNs;
  ShardID stays for the single-shard short form. Event carries the
  computed ShardID so consumers can route per-shard without rehashing.
- dispatcher.Pipeline gains Shards []int. When set, Run holds one
  Cursor + Schedule + Dispatcher per shard, opens one filer
  SubscribeMetadata stream with a predicate covering the whole set,
  and routes events into the matching shard's schedule from a single
  dispatch goroutine — no per-shard goroutine fan-out.
- shell command parses -shard or -shards (mutually exclusive),
  formats progress messages with a contiguous-range label when
  applicable, and validates against ShardCount.

Integration test now uses -shards 0-15 (one subprocess invocation)
instead of a 16-iteration loop.

* fix(s3/lifecycle): allow Reader with StartTsNs=0 + Cursor=nil

The reader rejected the legitimate 'fresh subscription from epoch'
state when called from a fresh Pipeline.Run on a multi-shard worker
(no cursor file yet, all shards' MinTsNs=0). The downstream
SubscribeMetadata call handles SinceNs=0 fine; the up-front check
was over-defensive and broke the auto-scheduler completely (CI
showed 5-second-cadence retries with this exact error).

* fix(s3/lifecycle): schedule from ModTime not eventTime

A backdated or out-of-band entry update has eventTime ≈ now while
ModTime is far in the past; eventTime+Delay would push the dispatch
into the future even though the rule already fires. ModTime+Delay
is the correct fire moment. The dispatcher's identity-CAS still
catches drift between schedule and dispatch.

* fix(s3/lifecycle): -runtime cap on run-shard so it exits on quiet shards

The CI integration test sets -events 200 expecting the subprocess to
return after 200 in-shard events. But -events counts only events that
pass the shard filter; the test produces ~5 such events (bucket
create, lifecycle PUT, two object PUTs, mtime backdate), so the
reader stays in stream.Recv forever and runShellCommand hangs the
test deadline.

- weed/shell/command_s3_lifecycle_run_shard.go: add -runtime D flag.
  When > 0, Pipeline.Run runs under context.WithTimeout(D); on
  expiry the reader/dispatcher drain cleanly and the cursor saves.
- weed/s3api/s3lifecycle/dispatcher/pipeline.go: treat
  context.DeadlineExceeded the same as context.Canceled at exit
  (both are graceful shutdown signals).

* test(s3/lifecycle): pass -runtime 10s to run-shard

Pair with the new -runtime flag so the subprocess exits cleanly
after 10s instead of waiting for an event budget that never lands
on quiet shards.

* refactor(s3/lifecycle): extract HashExtended to s3lifecycle pkg

The worker's router needs the same length-prefixed sha256 of the entry's
Extended map; pulling it out of the s3api private file lets both sides
import it.

* fix(s3/lifecycle): worker captures ExtendedHash for identity-CAS

Without this, the dispatcher sends ExpectedIdentity.ExtendedHash = nil
while the live entry on the server has a non-nil hash, so every dispatch
returns NOOP_RESOLVED:STALE_IDENTITY and nothing is ever deleted.

* fix(s3/lifecycle): identity HeadFid via GetFileIdString

Meta-log events go through BeforeEntrySerialization, which clears
FileChunk.FileId and writes the Fid struct instead. Reading .FileId
directly returns "" on the worker side while the server's freshly
fetched entry still has a populated string, so the identity-CAS would
mismatch and every expiration ended in NOOP_RESOLVED:STALE_IDENTITY.

* fix(s3/lifecycle): treat gRPC Canceled/DeadlineExceeded as graceful exit

errors.Is doesn't unwrap a gRPC status error back to the stdlib ctx
errors, so a subscription that ends because runCtx was canceled was
being logged as a fatal reader error. Check status.Code as well so the
shell's -runtime cap exits cleanly.

* fix(test/s3/lifecycle): pass the gRPC port (not HTTP) to run-shard

run-shard's -s3 flag dials the LifecycleDelete gRPC service, which
listens on s3.port + 10000. The integration test was passing the HTTP
port instead, so the dispatcher's RPC just timed out and the shell
command exited under -runtime with no work done.

* chore(test/s3/lifecycle): drop emoji from Makefile output

* docs(test/s3/lifecycle): correct '-shards 0-15' wording

* fix(s3/lifecycle): reject out-of-range shard IDs in Pipeline.Run

The shell's parseShardsSpec already validates, but a programmatic caller
(scheduler, future worker config) shouldn't be able to silently produce
no-op states by passing -1 or 99.

* fix(s3/lifecycle): bound drain + final-save with their own timeouts

Shutdown was using context.Background, so a stuck dispatcher RPC or
filer save could keep Pipeline.Run from ever returning.

* fix(test/s3/lifecycle): drop self-killing pkill in stop-server

The pkill pattern \"weed mini -dir=...\" is also in the running shell's
argv (it's the recipe body), so pkill -f matches its own bash and the
recipe exits with Terminated. CI test job passed but the cleanup step
failed with exit 2. The PID file is sufficient on its own.

* docs(test/s3/lifecycle): document S3_GRPC_ENDPOINT env var

.github/workflows/s3-go-tests.yml

@@ -245,6 +245,51 @@ jobs:
           path: test/s3/retention/weed-test*.log
           retention-days: 3
 
+  s3-lifecycle-tests:
+    name: S3 Lifecycle Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 15
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+        id: go
+
+      - name: Install SeaweedFS
+        run: |
+          go install -buildvcs=false
+
+      - name: Run S3 Lifecycle Tests
+        timeout-minutes: 12
+        working-directory: test/s3/lifecycle
+        run: |
+          set -x
+          make test-with-server
+
+      - name: Show server logs on failure
+        if: failure()
+        working-directory: test/s3/lifecycle
+        run: |
+          if [ -f weed-test.log ]; then
+            echo "=== Last 200 lines of server logs ==="
+            tail -200 weed-test.log
+          fi
+          ps aux | grep -E "(weed|test)" || true
+          netstat -tlnp 2>/dev/null | grep -E "(8333|9333|8080|8888)" || true
+
+      - name: Upload test logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: s3-lifecycle-test-logs
+          path: test/s3/lifecycle/weed-test*.log
+          retention-days: 3
+
   s3-checksum-tests:
     name: S3 Checksum Tests
     runs-on: ubuntu-22.04

test/s3/lifecycle/Makefile

@@ -0,0 +1,80 @@
+# S3 Lifecycle Test Makefile
+# End-to-end test of the event-driven lifecycle worker, driven by
+# the s3.lifecycle.run-shard shell command.
+
+.PHONY: help build-weed start-server stop-server test test-with-server clean health-check
+
+WEED_BINARY := ../../../weed/weed_binary
+S3_PORT := 8333
+MASTER_PORT := 9333
+VOLUME_PORT := 8080
+FILER_PORT := 8888
+ACCESS_KEY ?= some_access_key1
+SECRET_KEY ?= some_secret_key1
+TEST_TIMEOUT := 10m
+TEST_PATTERN := TestLifecycle
+SERVER_DIR := ./test-volume-data/server-data
+
+help:
+	@echo "S3 Lifecycle Test Makefile"
+	@echo ""
+	@echo "Targets:"
+	@echo "  build-weed       - build the SeaweedFS binary"
+	@echo "  start-server     - start a local 'weed mini' cluster"
+	@echo "  stop-server      - stop the cluster"
+	@echo "  test             - run tests against an already-running cluster"
+	@echo "  test-with-server - start, run tests, stop"
+	@echo "  health-check     - is the S3 endpoint up?"
+	@echo "  clean            - remove test artifacts"
+
+build-weed:
+	@echo "Building SeaweedFS binary..."
+	@cd ../../../weed && go build -o weed_binary .
+	@chmod +x $(WEED_BINARY)
+
+start-server: build-weed
+	@echo "Starting weed mini..."
+	@rm -f weed-server.pid
+	@mkdir -p $(SERVER_DIR)
+	@AWS_ACCESS_KEY_ID=$(ACCESS_KEY) AWS_SECRET_ACCESS_KEY=$(SECRET_KEY) $(WEED_BINARY) mini \
+		-dir=$(SERVER_DIR) \
+		-s3.port=$(S3_PORT) \
+		> weed-test.log 2>&1 & \
+		echo $$! > weed-server.pid
+	@for i in $$(seq 1 90); do \
+		if curl -s http://localhost:$(S3_PORT) >/dev/null 2>&1; then \
+			echo "server up after $$i s"; exit 0; \
+		fi; \
+		sleep 1; \
+	done; \
+	echo "server did not start within 90s"; \
+	tail -50 weed-test.log; \
+	exit 1
+
+stop-server:
+	@if [ -f weed-server.pid ]; then \
+		PID=$$(cat weed-server.pid); \
+		kill -TERM $$PID 2>/dev/null || true; \
+		sleep 1; \
+		kill -KILL $$PID 2>/dev/null || true; \
+		rm -f weed-server.pid; \
+	fi
+
+health-check:
+	@curl -s http://localhost:$(S3_PORT) >/dev/null && echo "S3 up on $(S3_PORT)" || (echo "S3 not reachable"; exit 1)
+
+test:
+	@WEED_BINARY=$$(cd ../../../weed && pwd)/weed_binary \
+	S3_ENDPOINT=http://localhost:$(S3_PORT) \
+	S3_GRPC_ENDPOINT=localhost:$$(($(S3_PORT) + 10000)) \
+	MASTER_ENDPOINT=http://localhost:$(MASTER_PORT) \
+	FILER_GRPC_ADDRESS=localhost:$$(($(FILER_PORT) + 10000)) \
+	go test -v -timeout $(TEST_TIMEOUT) -run $(TEST_PATTERN)
+
+test-with-server: start-server
+	@$(MAKE) test || (RC=$$?; $(MAKE) stop-server; exit $$RC)
+	@$(MAKE) stop-server
+
+clean: stop-server
+	@rm -rf $(SERVER_DIR) test-volume-data
+	@rm -f weed-test.log weed-server.pid

test/s3/lifecycle/README.md

@@ -0,0 +1,38 @@
+# S3 Lifecycle Integration Tests
+
+End-to-end test of the event-driven S3 lifecycle worker, exercised through
+the `s3.lifecycle.run-shard` shell command.
+
+## Why backdate mtimes?
+
+The S3 API rejects `Expiration.Days < 1`, so a literal "wait one day"
+integration test isn't workable. Each test sets up a 1-day expiration rule,
+puts the target object, then rewrites its filer entry's `Mtime` to ~30 days
+ago via `filer_pb.UpdateEntry`. From the engine's perspective the object is
+past its expiration window the moment the shell command starts.
+
+## Running
+
+```sh
+# build the binary, start a local mini cluster, run tests, stop it
+make test-with-server
+
+# or, if a cluster is already running on the default ports
+make test
+```
+
+The test runs the shell command once with `-shards 0-15` (one filer
+subscription covering all 16 shards) rather than computing the target
+object's shard up front. This keeps the test independent of the
+`ShardID(bucket, key)` hash function — only that some shard reaches the
+deletion within the polling window.
+
+## Environment
+
+| variable             | default                  | description                |
+|----------------------|--------------------------|----------------------------|
+| `WEED_BINARY`        | _required_               | path to `weed_binary`      |
+| `S3_ENDPOINT`        | `http://localhost:8333`  | S3 API URL                 |
+| `S3_GRPC_ENDPOINT`   | `localhost:18333`        | S3 gRPC for lifecycle dispatch |
+| `MASTER_ENDPOINT`    | `http://localhost:9333`  | master HTTP                |
+| `FILER_GRPC_ADDRESS` | `localhost:18888`        | filer gRPC for `UpdateEntry` |

test/s3/lifecycle/s3_lifecycle_test.go

@@ -0,0 +1,233 @@
+// Package lifecycle is the end-to-end test for the event-driven S3
+// lifecycle worker, driven by the s3.lifecycle.run-shard shell command.
+//
+// The S3 API rejects Expiration.Days < 1, so a literal "wait one day"
+// integration test is unworkable. Instead, the test backdates the target
+// object's mtime via the filer's UpdateEntry RPC: from the engine's
+// perspective the object is past its expiration window the moment the
+// shell command starts, and the dispatcher fires immediately.
+//
+// Variables (set by the Makefile):
+//
+//	WEED_BINARY         - path to the built `weed` binary
+//	S3_ENDPOINT         - http://host:port for the S3 API
+//	FILER_GRPC_ADDRESS  - host:port of the filer's gRPC listener
+package lifecycle
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/url"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"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"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+const (
+	defaultS3Endpoint     = "http://localhost:8333"
+	defaultS3GrpcEndpoint = "localhost:18333"
+	defaultMasterEndpt    = "http://localhost:9333"
+	defaultFilerGRPC      = "localhost:18888"
+	bucketLifecycleXMLKey = "s3-bucket-lifecycle-configuration-xml"
+	bucketsPath           = "/buckets"
+	accessKey             = "some_access_key1"
+	secretKey           = "some_secret_key1"
+	region              = "us-east-1"
+)
+
+func envOr(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+func s3Client(t *testing.T) *s3.Client {
+	t.Helper()
+	endpoint := envOr("S3_ENDPOINT", defaultS3Endpoint)
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion(region),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")),
+		config.WithEndpointResolverWithOptions(aws.EndpointResolverWithOptionsFunc(
+			func(service, region string, _ ...interface{}) (aws.Endpoint, error) {
+				return aws.Endpoint{URL: endpoint, SigningRegion: region, HostnameImmutable: true}, nil
+			})),
+	)
+	require.NoError(t, err)
+	return s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = true })
+}
+
+func filerClient(t *testing.T) (filer_pb.SeaweedFilerClient, func()) {
+	t.Helper()
+	addr := envOr("FILER_GRPC_ADDRESS", defaultFilerGRPC)
+	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	return filer_pb.NewSeaweedFilerClient(conn), func() { conn.Close() }
+}
+
+// uniqueBucket returns a fresh bucket name; the shell command picks up all
+// buckets with lifecycle config, so two parallel-running tests would conflict
+// on cursor state. Each test gets its own bucket.
+func uniqueBucket(prefix string) string {
+	return fmt.Sprintf("lc-%s-%d", prefix, time.Now().UnixNano())
+}
+
+func mustCreateBucket(t *testing.T, c *s3.Client, name string) {
+	t.Helper()
+	_, 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)})
+		if listOut != nil {
+			for _, o := range listOut.Contents {
+				c.DeleteObject(context.Background(), &s3.DeleteObjectInput{Bucket: aws.String(name), Key: o.Key})
+			}
+		}
+		c.DeleteBucket(context.Background(), &s3.DeleteBucketInput{Bucket: aws.String(name)})
+	})
+}
+
+func putExpirationLifecycle(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-prefix"),
+					Status: types.ExpirationStatusEnabled,
+					Filter: &types.LifecycleRuleFilter{Prefix: aws.String(prefix)},
+					Expiration: &types.LifecycleExpiration{
+						Days: aws.Int32(days),
+					},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+}
+
+func putObject(t *testing.T, c *s3.Client, bucket, key, body string) {
+	t.Helper()
+	_, err := c.PutObject(context.Background(), &s3.PutObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+		Body:   strings.NewReader(body),
+	})
+	require.NoError(t, err)
+}
+
+// backdateMtime rewrites the object's filer entry attributes so its Mtime
+// is daysOld days in the past. This sidesteps the AWS-spec minimum of one
+// day for Expiration.Days, letting the lifecycle dispatcher fire on demand.
+func backdateMtime(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket, key string, daysOld int) {
+	t.Helper()
+	dir, name := splitBucketKey(bucket, key)
+	resp, err := fc.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+		Directory: dir,
+		Name:      name,
+	})
+	require.NoError(t, err, "lookup %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, "update %s/%s", dir, name)
+}
+
+func splitBucketKey(bucket, key string) (dir, name string) {
+	full := bucketsPath + "/" + bucket + "/" + key
+	if i := strings.LastIndex(full, "/"); i >= 0 {
+		return full[:i], full[i+1:]
+	}
+	return full, ""
+}
+
+// runShellCommand invokes `weed shell` with a one-shot s3.lifecycle.run-shard
+// command piped via stdin and returns the combined stdout+stderr.
+func runShellCommand(t *testing.T, command string) string {
+	t.Helper()
+	binary := envOr("WEED_BINARY", "")
+	require.NotEmpty(t, binary, "WEED_BINARY must be set")
+	masterEndpoint := envOr("MASTER_ENDPOINT", defaultMasterEndpt)
+	master := strings.TrimPrefix(masterEndpoint, "http://")
+	if u, err := url.Parse(masterEndpoint); err == nil && u.Host != "" {
+		master = u.Host
+	}
+	cmd := exec.Command(binary, "shell", "-master="+master)
+	cmd.Stdin = strings.NewReader(command + "\nexit\n")
+	var out bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &out, &out
+	err := cmd.Run()
+	output := out.String()
+	if err != nil {
+		t.Logf("shell command output:\n%s", output)
+		t.Fatalf("shell exec failed: %v", err)
+	}
+	return output
+}
+
+// TestLifecycleExpirationFiresOnBackdatedObject is the end-to-end
+// validation: a 1-day expiration rule plus an object whose mtime has been
+// backdated to 30 days ago must result in deletion when the shell command
+// runs the matching shard.
+func TestLifecycleExpirationFiresOnBackdatedObject(t *testing.T) {
+	c := s3Client(t)
+	fc, fcClose := filerClient(t)
+	defer fcClose()
+
+	bucket := uniqueBucket("expire")
+	mustCreateBucket(t, c, bucket)
+	putExpirationLifecycle(t, c, bucket, "expire/", 1)
+
+	const oldKey = "expire/old.txt"
+	const freshKey = "keep/fresh.txt"
+	putObject(t, c, bucket, oldKey, "old")
+	putObject(t, c, bucket, freshKey, "fresh")
+	backdateMtime(t, fc, bucket, oldKey, 30)
+
+	// One subscription handles every shard via -shards 0-15; this stays
+	// independent of which (bucket, key) hash lands the target on.
+	// -runtime caps the run by wall-clock so the subprocess exits even
+	// when the in-shard event count never reaches -events.
+	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),
+	))
+	t.Logf("shell command output:\n%s", out)
+	require.NotContains(t, out, "FATAL", "shell output:\n%s", out)
+
+	// Allow the dispatched delete to land in the filer + propagate to S3.
+	require.Eventuallyf(t, func() bool {
+		_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket), Key: aws.String(oldKey),
+		})
+		return err != nil
+	}, 30*time.Second, 500*time.Millisecond, "expected %s/%s to be deleted", bucket, oldKey)
+
+	// The fresh, in-prefix-but-recent object must remain (1d rule, mtime
+	// is now). Best-evidence check after waiting for the same poll window.
+	_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
+		Bucket: aws.String(bucket), Key: aws.String(freshKey),
+	})
+	require.NoError(t, err, "%s should still exist", freshKey)
+}

weed/s3api/s3api_internal_lifecycle.go

@@ -3,15 +3,13 @@ package s3api
 import (
 	"bytes"
 	"context"
-	"crypto/sha256"
 	"errors"
-	"sort"
-	"strconv"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"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"
 )
 
 // LifecycleDelete executes one (rule, action) verdict: re-fetch, identity
@@ -36,8 +34,7 @@ func (s3a *S3ApiServer) LifecycleDelete(ctx context.Context, req *s3_lifecycle_p
 		return retryLater("TRANSPORT_ERROR: " + err.Error()), nil
 	}
 
-	live := computeEntryIdentity(entry)
-	if !identityMatches(live, req.ExpectedIdentity) {
+	if !identityMatches(computeEntryIdentity(entry), req.ExpectedIdentity) {
 		return noopResolved("STALE_IDENTITY"), nil
 	}
 
@@ -142,36 +139,12 @@ func computeEntryIdentity(entry *filer_pb.Entry) *s3_lifecycle_pb.EntryIdentity
 		id.Size = int64(entry.Attributes.FileSize)
 	}
 	if len(entry.GetChunks()) > 0 {
-		id.HeadFid = entry.GetChunks()[0].FileId
+		id.HeadFid = entry.GetChunks()[0].GetFileIdString()
 	}
-	id.ExtendedHash = hashExtended(entry.Extended)
+	id.ExtendedHash = s3lifecycle.HashExtended(entry.Extended)
 	return id
 }
 
-// hashExtended is length-prefixed; a forged tag value can't collide with a
-// legitimate multi-tag map.
-func hashExtended(ext map[string][]byte) []byte {
-	if len(ext) == 0 {
-		return nil
-	}
-	keys := make([]string, 0, len(ext))
-	for k := range ext {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-	h := sha256.New()
-	for _, k := range keys {
-		h.Write([]byte(strconv.Itoa(len(k))))
-		h.Write([]byte{':'})
-		h.Write([]byte(k))
-		v := ext[k]
-		h.Write([]byte(strconv.Itoa(len(v))))
-		h.Write([]byte{':'})
-		h.Write(v)
-	}
-	return h.Sum(nil)
-}
-
 func identityMatches(live, want *s3_lifecycle_pb.EntryIdentity) bool {
 	if want == nil {
 		// No CAS witness (early bootstrap); skip.

weed/s3api/s3api_internal_lifecycle_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 )
 
 func TestComputeEntryIdentity_BasicFields(t *testing.T) {
@@ -45,7 +46,7 @@ func TestComputeEntryIdentity_NilSafeMissingChunks(t *testing.T) {
 func TestHashExtended_OrderStable(t *testing.T) {
 	a := map[string][]byte{"k1": []byte("v1"), "k2": []byte("v2")}
 	b := map[string][]byte{"k2": []byte("v2"), "k1": []byte("v1")}
-	if !bytes.Equal(hashExtended(a), hashExtended(b)) {
+	if !bytes.Equal(s3lifecycle.HashExtended(a), s3lifecycle.HashExtended(b)) {
 		t.Fatalf("hash should be insensitive to map iteration order")
 	}
 }
@@ -55,16 +56,16 @@ func TestHashExtended_DelimiterCollisionResistant(t *testing.T) {
 	// Length-prefix encoding must keep them apart.
 	a := map[string][]byte{"k1": []byte("v1"), "k2": []byte("v2")}
 	b := map[string][]byte{"k1": []byte("v1k2v2")}
-	if bytes.Equal(hashExtended(a), hashExtended(b)) {
+	if bytes.Equal(s3lifecycle.HashExtended(a), s3lifecycle.HashExtended(b)) {
 		t.Fatalf("delimiter-forged Extended payloads must not collide")
 	}
 }
 
 func TestHashExtended_NilEqualsEmpty(t *testing.T) {
-	if got := hashExtended(nil); len(got) != 0 {
+	if got := s3lifecycle.HashExtended(nil); len(got) != 0 {
 		t.Fatalf("nil should produce zero-length hash, got %d bytes", len(got))
 	}
-	if got := hashExtended(map[string][]byte{}); len(got) != 0 {
+	if got := s3lifecycle.HashExtended(map[string][]byte{}); len(got) != 0 {
 		t.Fatalf("empty map should produce zero-length hash, got %d bytes", len(got))
 	}
 }

weed/s3api/s3lifecycle/dispatcher/dispatcher.go

@@ -200,8 +200,9 @@ func toProtoIdentity(id *router.EntryIdentity) *s3_lifecycle_pb.EntryIdentity {
 		return nil
 	}
 	return &s3_lifecycle_pb.EntryIdentity{
-		MtimeNs: id.MtimeNs,
-		Size:    id.Size,
-		HeadFid: id.HeadFid,
+		MtimeNs:      id.MtimeNs,
+		Size:         id.Size,
+		HeadFid:      id.HeadFid,
+		ExtendedHash: id.ExtendedHash,
 	}
 }

weed/s3api/s3lifecycle/dispatcher/pipeline.go

@@ -9,18 +9,31 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
-// Pipeline composes the per-shard reader, router, dispatcher, and cursor
-// checkpoint into a single Run loop. One Pipeline per (worker, shard).
+// Pipeline composes the reader, router, dispatcher, and cursor checkpoint
+// into a single Run loop. One Pipeline can handle a contiguous shard span
+// or any explicit set of shards via Shards; ShardID still works for the
+// single-shard case (and is preferred for short-form configuration).
+//
+// Internally there is exactly one filer subscription regardless of how
+// many shards Shards contains; events are filtered by the reader's
+// ShardPredicate and routed to the matching shard's Cursor + Schedule
+// inside the existing dispatch goroutine — no per-shard goroutines.
 type Pipeline struct {
-	ShardID     int
+	ShardID     int   // used when Shards is empty
+	Shards      []int // overrides ShardID when non-empty
 	BucketsPath string
 
-	Engine    *engine.Engine
+	Engine *engine.Engine
+	// Cursor is consulted only when len(Shards) <= 1. Range mode allocates
+	// a fresh Cursor per shard internally.
 	Cursor    *reader.Cursor
 	Persister reader.Persister
 	Client    LifecycleClient
@@ -47,46 +60,88 @@ const (
 	defaultDispatchTick   = 5 * time.Second
 	defaultCheckpointTick = 30 * time.Second
 	defaultEventBuffer    = 1024
+	shutdownDrainTimeout  = 30 * time.Second
+	shutdownSaveTimeout   = 5 * time.Second
 )
 
-// Run blocks until ctx is canceled or a fatal error occurs. On exit, the
-// cursor is persisted; in-flight schedule entries are dropped (the meta-log
-// is the durable buffer, so a restart re-derives them).
+// shardState bundles per-shard mutable state so the single dispatch
+// goroutine can route an event to the right cursor + schedule by lookup.
+type shardState struct {
+	cursor   *reader.Cursor
+	dispatch *Dispatcher
+}
+
+// Run blocks until ctx is canceled or a fatal error occurs. On exit, every
+// shard's cursor is persisted; in-flight schedule entries are dropped
+// (the meta-log is the durable buffer, so a restart re-derives them).
 func (p *Pipeline) Run(ctx context.Context) error {
-	if p.Engine == nil || p.Cursor == nil || p.Persister == nil ||
-		p.Client == nil || p.FilerClient == nil {
+	if p.Engine == nil || p.Persister == nil || p.Client == nil || p.FilerClient == nil {
 		return errors.New("pipeline: missing required dependency")
 	}
 	if p.BucketsPath == "" {
 		return errors.New("pipeline: BucketsPath required")
 	}
 
-	// Restore cursor; freezes re-arm naturally when the reader re-encounters
-	// the poison event at MinTsNs and the dispatch state machine drives it
+	// Resolve the active shard set. Single-shard configurations populate
+	// either Shards=[N] or Shards=nil with ShardID=N (latter is the legacy
+	// path that also supplies a Cursor); both feed the same range model.
+	shardIDs := p.Shards
+	if len(shardIDs) == 0 {
+		shardIDs = []int{p.ShardID}
+	}
+	shardSet := make(map[int]struct{}, len(shardIDs))
+	for _, s := range shardIDs {
+		if s < 0 || s >= s3lifecycle.ShardCount {
+			return fmt.Errorf("pipeline: shard %d out of [0,%d)", s, s3lifecycle.ShardCount)
+		}
+		shardSet[s] = struct{}{}
+	}
+
+	// Per-shard cursor + dispatcher. Cursors restore from the durable
+	// store; freezes re-arm naturally when the reader re-encounters the
+	// poison event at MinTsNs and the dispatch state machine drives it
 	// back to BLOCKED.
-	state, err := p.Persister.Load(ctx, p.ShardID)
-	if err != nil {
-		return fmt.Errorf("cursor load: %w", err)
+	states := make(map[int]*shardState, len(shardIDs))
+	var minStartTsNs int64 = -1
+	for _, shardID := range shardIDs {
+		c := p.Cursor
+		if len(shardIDs) != 1 || c == nil {
+			c = reader.NewCursor()
+		}
+		state, err := p.Persister.Load(ctx, shardID)
+		if err != nil {
+			return fmt.Errorf("cursor load shard=%d: %w", shardID, err)
+		}
+		c.Restore(state)
+		states[shardID] = &shardState{
+			cursor: c,
+			dispatch: &Dispatcher{
+				ShardID:  shardID,
+				Client:   p.Client,
+				Cursor:   c,
+				Schedule: router.NewSchedule(),
+			},
+		}
+		if mt := c.MinTsNs(); mt > 0 && (minStartTsNs < 0 || mt < minStartTsNs) {
+			minStartTsNs = mt
+		}
 	}
-	p.Cursor.Restore(state)
-
-	dispatch := &Dispatcher{
-		ShardID:  p.ShardID,
-		Client:   p.Client,
-		Cursor:   p.Cursor,
-		Schedule: router.NewSchedule(),
+	if minStartTsNs < 0 {
+		minStartTsNs = 0
 	}
 
-	// 2. Wire reader -> router -> schedule via a buffered channel.
 	bufSize := p.EventBuffer
 	if bufSize <= 0 {
 		bufSize = defaultEventBuffer
 	}
 	events := make(chan *reader.Event, bufSize)
 	rd := &reader.Reader{
-		ShardID:     p.ShardID,
 		BucketsPath: p.BucketsPath,
-		Cursor:      p.Cursor,
+		ShardPredicate: func(s int) bool {
+			_, ok := shardSet[s]
+			return ok
+		},
+		StartTsNs:   minStartTsNs,
 		Events:      events,
 		EventBudget: p.EventBudget,
 	}
@@ -104,13 +159,16 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		defer wg.Done()
 		defer close(events)
 		readerErr = rd.Run(runCtx, p.FilerClient, p.ClientName, p.ClientID)
-		if readerErr != nil && !errors.Is(readerErr, context.Canceled) {
-			glog.Errorf("lifecycle reader: shard=%d: %v", p.ShardID, readerErr)
+		if readerErr != nil && !isCtxShutdown(readerErr) {
+			glog.Errorf("lifecycle reader: shards=%v: %v", shardIDs, readerErr)
 		}
 		cancel() // wake the dispatcher goroutine to drain & exit
 	}()
 
-	// Router/dispatcher goroutine: pulls events, routes them, ticks schedule.
+	// Router/dispatcher goroutine: pulls events, routes them to per-shard
+	// schedules, ticks every shard's dispatcher on the same cadence, and
+	// checkpoints every shard's cursor on the checkpoint cadence. One
+	// goroutine handles all shards — there is no fan-out per shard.
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -128,28 +186,46 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		defer ct.Stop()
 		snap := p.Engine.Snapshot()
 
+		drainAll := func() {
+			drainCtx, drainCancel := context.WithTimeout(context.Background(), shutdownDrainTimeout)
+			defer drainCancel()
+			now := time.Now()
+			for _, st := range states {
+				st.dispatch.Tick(drainCtx, now)
+			}
+		}
+
 		for {
 			select {
 			case <-runCtx.Done():
-				dispatch.Tick(context.Background(), time.Now())
+				drainAll()
 				return
 			case ev, ok := <-events:
 				if !ok {
-					dispatch.Tick(context.Background(), time.Now())
+					drainAll()
 					return
 				}
+				st := states[ev.ShardID]
+				if st == nil {
+					continue
+				}
 				if snap == nil {
 					snap = p.Engine.Snapshot()
 				}
 				for _, m := range router.Route(snap, ev, time.Now()) {
-					dispatch.Schedule.Add(m)
+					st.dispatch.Schedule.Add(m)
 				}
 			case <-dt.C:
-				snap = p.Engine.Snapshot() // refresh between tick boundaries
-				dispatch.Tick(runCtx, time.Now())
+				snap = p.Engine.Snapshot()
+				now := time.Now()
+				for _, st := range states {
+					st.dispatch.Tick(runCtx, now)
+				}
 			case <-ct.C:
-				if err := p.Persister.Save(runCtx, p.ShardID, p.Cursor.Snapshot()); err != nil {
-					glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", p.ShardID, err)
+				for shardID, st := range states {
+					if err := p.Persister.Save(runCtx, shardID, st.cursor.Snapshot()); err != nil {
+						glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", shardID, err)
+					}
 				}
 			}
 		}
@@ -158,12 +234,33 @@ func (p *Pipeline) Run(ctx context.Context) error {
 	wg.Wait()
 
 	// Final cursor checkpoint on graceful shutdown.
-	if err := p.Persister.Save(context.Background(), p.ShardID, p.Cursor.Snapshot()); err != nil {
-		glog.Warningf("lifecycle cursor final save: shard=%d: %v", p.ShardID, err)
+	for shardID, st := range states {
+		saveCtx, saveCancel := context.WithTimeout(context.Background(), shutdownSaveTimeout)
+		err := p.Persister.Save(saveCtx, shardID, st.cursor.Snapshot())
+		saveCancel()
+		if err != nil {
+			glog.Warningf("lifecycle cursor final save: shard=%d: %v", shardID, err)
+		}
 	}
 
-	if readerErr != nil && !errors.Is(readerErr, context.Canceled) {
+	if readerErr != nil && !isCtxShutdown(readerErr) {
 		return readerErr
 	}
 	return nil
 }
+
+// isCtxShutdown reports whether err is a graceful ctx-driven shutdown
+// (Canceled or DeadlineExceeded), including the gRPC status forms that
+// don't unwrap to the std-lib ctx errors.
+func isCtxShutdown(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	if code := status.Code(err); code == codes.Canceled || code == codes.DeadlineExceeded {
+		return true
+	}
+	return false
+}

weed/s3api/s3lifecycle/identity.go

@@ -0,0 +1,36 @@
+package s3lifecycle
+
+import (
+	"crypto/sha256"
+	"sort"
+	"strconv"
+)
+
+// HashExtended is the canonical fingerprint of an entry's Extended map for
+// use in lifecycle identity-CAS. Length-prefixed so a forged tag value
+// can't collide with a legitimate multi-tag map. Returns nil for empty.
+//
+// Both sides of the LifecycleDelete CAS — the worker that captures the
+// schedule-time identity and the server that re-fetches the live entry —
+// must call this same function so the bytes match exactly.
+func HashExtended(ext map[string][]byte) []byte {
+	if len(ext) == 0 {
+		return nil
+	}
+	keys := make([]string, 0, len(ext))
+	for k := range ext {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	h := sha256.New()
+	for _, k := range keys {
+		h.Write([]byte(strconv.Itoa(len(k))))
+		h.Write([]byte{':'})
+		h.Write([]byte(k))
+		v := ext[k]
+		h.Write([]byte(strconv.Itoa(len(v))))
+		h.Write([]byte{':'})
+		h.Write(v)
+	}
+	return h.Sum(nil)
+}

weed/s3api/s3lifecycle/reader/reader.go

@@ -17,6 +17,7 @@ type Event struct {
 	TsNs      int64
 	Bucket    string
 	Key       string
+	ShardID   int
 	OldEntry  *filer_pb.Entry
 	NewEntry  *filer_pb.Entry
 	NewParent string
@@ -32,15 +33,24 @@ func (e *Event) IsCreate() bool {
 	return e.OldEntry == nil && e.NewEntry != nil
 }
 
-// Reader subscribes to the filer meta-log for one shard. It owns a Cursor
-// (per-ActionKey position within the shard) and emits in-shard Events to a
-// channel; the downstream router consumes events and ack-advances the cursor
-// for matched ActionKeys when their actions complete.
+// Reader subscribes to the filer meta-log and emits in-range Events to a
+// channel. One subscription handles a contiguous span (or arbitrary set)
+// of shards via ShardPredicate; the downstream router/dispatcher consume
+// events and ack-advance the per-shard cursor for matched ActionKeys
+// when their actions complete.
 type Reader struct {
-	ShardID     int    // [0, s3lifecycle.ShardCount)
+	// ShardID and ShardPredicate are alternatives — set at most one.
+	// ShardPredicate wins if both are populated.
+	ShardID        int             // [0, s3lifecycle.ShardCount); used when ShardPredicate is nil
+	ShardPredicate func(int) bool  // accepts an event when true; nil falls back to ShardID equality
+
 	BucketsPath string // e.g. "/buckets"
-	Cursor      *Cursor
-	Events      chan<- *Event
+	// Cursor is the single-shard cursor used for SinceNs when StartTsNs is 0.
+	// Range callers pass StartTsNs directly and leave Cursor nil; SinceNs=0
+	// then means "subscribe from the start of the meta-log".
+	Cursor    *Cursor
+	StartTsNs int64
+	Events    chan<- *Event
 
 	// EventBudget caps how many events Run processes before returning nil.
 	// Zero = unbounded; the run continues until ctx cancellation or stream
@@ -53,15 +63,15 @@ type Reader struct {
 	bucketsPathSlash string
 }
 
-// Run subscribes via SubscribeMetadata starting at Cursor.MinTsNs(), filters
-// to this shard, and emits Events. Returns on ctx.Done(), io.EOF, or
-// stream error. Caller is responsible for closing Events if it owns it.
+// Run subscribes via SubscribeMetadata starting at the configured position,
+// filters to the configured shard set, and emits Events. Returns on
+// ctx.Done(), io.EOF, or stream error. Caller is responsible for closing
+// Events if it owns it.
 func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, clientName string, clientID int32) error {
-	if r.ShardID < 0 || r.ShardID >= s3lifecycle.ShardCount {
-		return fmt.Errorf("reader: shard_id %d out of range", r.ShardID)
-	}
-	if r.Cursor == nil {
-		return errors.New("reader: nil Cursor")
+	if r.ShardPredicate == nil {
+		if r.ShardID < 0 || r.ShardID >= s3lifecycle.ShardCount {
+			return fmt.Errorf("reader: shard_id %d out of range and no ShardPredicate", r.ShardID)
+		}
 	}
 	if r.Events == nil {
 		return errors.New("reader: nil Events channel")
@@ -74,10 +84,14 @@ func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, cl
 		r.bucketsPathSlash += "/"
 	}
 
+	sinceNs := r.StartTsNs
+	if sinceNs == 0 && r.Cursor != nil {
+		sinceNs = r.Cursor.MinTsNs()
+	}
 	stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
 		ClientName:             clientName,
 		PathPrefix:             r.BucketsPath,
-		SinceNs:                r.Cursor.MinTsNs(),
+		SinceNs:                sinceNs,
 		ClientId:               clientID,
 		ClientSupportsBatching: true,
 	})
@@ -118,7 +132,12 @@ func (r *Reader) dispatchOne(ctx context.Context, resp *filer_pb.SubscribeMetada
 	if !ok {
 		return nil
 	}
-	if s3lifecycle.ShardID(bucket, key) != r.ShardID {
+	shardID := s3lifecycle.ShardID(bucket, key)
+	if r.ShardPredicate != nil {
+		if !r.ShardPredicate(shardID) {
+			return nil
+		}
+	} else if shardID != r.ShardID {
 		return nil
 	}
 
@@ -126,6 +145,7 @@ func (r *Reader) dispatchOne(ctx context.Context, resp *filer_pb.SubscribeMetada
 		TsNs:      resp.TsNs,
 		Bucket:    bucket,
 		Key:       key,
+		ShardID:   shardID,
 		OldEntry:  resp.EventNotification.OldEntry,
 		NewEntry:  resp.EventNotification.NewEntry,
 		NewParent: resp.EventNotification.NewParentPath,
@@ -212,6 +232,14 @@ func (r *Reader) extractBucketKey(resp *filer_pb.SubscribeMetadataResponse) (str
 // LogStartup is a small helper for callers that want a one-line readable
 // description of where the reader is starting.
 func (r *Reader) LogStartup() {
+	sinceNs := r.StartTsNs
+	if sinceNs == 0 && r.Cursor != nil {
+		sinceNs = r.Cursor.MinTsNs()
+	}
+	if r.ShardPredicate != nil {
+		glog.V(1).Infof("lifecycle reader: shard=range sinceNs=%d budget=%d", sinceNs, r.EventBudget)
+		return
+	}
 	glog.V(1).Infof("lifecycle reader: shard=%d sinceNs=%d budget=%d",
-		r.ShardID, r.Cursor.MinTsNs(), r.EventBudget)
+		r.ShardID, sinceNs, r.EventBudget)
 }

weed/s3api/s3lifecycle/router/router.go

@@ -31,9 +31,10 @@ type Match struct {
 // s3_lifecycle_pb.EntryIdentity but stay in-package so the router doesn't
 // pull a proto dependency.
 type EntryIdentity struct {
-	MtimeNs int64
-	Size    int64
-	HeadFid string
+	MtimeNs      int64
+	Size         int64
+	HeadFid      string
+	ExtendedHash []byte
 }
 
 // Route returns the matches that fire for ev against snap. Only EVENT_DRIVEN
@@ -70,11 +71,13 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match {
 		if action.Mode != engine.ModeEventDriven {
 			continue
 		}
-		// Evaluate at the scheduled dispatch time, not the event time:
-		// ExpirationDays gates on now >= modtime + N, which is exactly
-		// what holds when we actually dispatch. The dispatcher's
-		// identity-CAS catches drift if the object changes meanwhile.
-		dueTime := eventTime.Add(action.Delay)
+		// 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)
 		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
 		if res.Action == s3lifecycle.ActionNone {
 			continue
@@ -138,8 +141,12 @@ func buildIdentity(ev *reader.Event) *EntryIdentity {
 		id.Size = int64(entry.Attributes.FileSize)
 	}
 	if len(entry.GetChunks()) > 0 {
-		id.HeadFid = entry.GetChunks()[0].FileId
+		// Meta-log events arrive with chunk.FileId cleared by
+		// BeforeEntrySerialization; GetFileIdString reconstructs it from
+		// Fid so the worker matches the server-side fingerprint.
+		id.HeadFid = entry.GetChunks()[0].GetFileIdString()
 	}
+	id.ExtendedHash = s3lifecycle.HashExtended(entry.Extended)
 	return id
 }
 

weed/s3api/s3lifecycle/router/router_test.go

@@ -195,3 +195,34 @@ func TestRouteIdentityCapturedForNewEntry(t *testing.T) {
 		t.Fatalf("MtimeNs=%d, want %d (Mtime*1e9)", id.MtimeNs, wantNs)
 	}
 }
+
+func TestRouteIdentityHashesExtended(t *testing.T) {
+	// A normal S3 PUT stores ETag/content-type etc. in Extended; the worker
+	// must hash these into ExtendedHash so the server's identity-CAS sees
+	// the same fingerprint. Without this the live entry's non-nil
+	// ExtendedHash diverges from the worker's nil value and every
+	// dispatch returns NOOP_RESOLVED:STALE_IDENTITY.
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWith(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.Add(-48 * time.Hour)
+	ev := eventCreate("bk", "k", old.Unix(), 1, old.UnixNano())
+	ev.NewEntry.Extended = map[string][]byte{
+		"X-Amz-Meta-Etag": []byte("abc123"),
+		"Content-Type":    []byte("text/plain"),
+	}
+
+	matches := Route(snap, ev, now)
+	if len(matches) != 1 {
+		t.Fatalf("expected 1 match, got %v", matches)
+	}
+	id := matches[0].Identity
+	if id == nil || len(id.ExtendedHash) == 0 {
+		t.Fatalf("ExtendedHash not captured for non-empty Extended: %+v", id)
+	}
+	want := s3lifecycle.HashExtended(ev.NewEntry.Extended)
+	if string(id.ExtendedHash) != string(want) {
+		t.Fatalf("ExtendedHash mismatch:\n got %x\nwant %x", id.ExtendedHash, want)
+	}
+}

weed/shell/command_s3_lifecycle_run_shard.go

@@ -0,0 +1,376 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/lifecycle_xml"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3LifecycleRunShard{})
+}
+
+type commandS3LifecycleRunShard struct{}
+
+func (c *commandS3LifecycleRunShard) Name() string {
+	return "s3.lifecycle.run-shard"
+}
+
+func (c *commandS3LifecycleRunShard) Help() string {
+	return `manually run one or more shards of the event-driven S3 lifecycle worker
+
+Subscribes once to the filer meta-log, filters events to the configured
+(bucket, key-prefix-hash) shards, routes them through the compiled lifecycle
+engine, and dispatches due actions to the S3 server's LifecycleDelete RPC.
+Persists each shard's cursor to /etc/s3/lifecycle/cursors/shard-NN.json so
+subsequent runs resume.
+
+The -shards form covers a range or set; one filer subscription handles the
+whole set, with no per-shard goroutine fan-out. Provide either -shard or
+-shards, not both.
+
+	# single shard
+	s3.lifecycle.run-shard -shard 0 -s3 localhost:8333 -events 100
+
+	# contiguous range, all 16 shards via one subscription
+	s3.lifecycle.run-shard -shards 0-15 -s3 localhost:8333 -events 5000
+
+	# explicit set
+	s3.lifecycle.run-shard -shards 0,3,7 -s3 localhost:8333
+
+	# custom cadence
+	s3.lifecycle.run-shard -shards 0-15 -s3 s3-host:8333 -dispatch 1s -checkpoint 10s
+`
+}
+
+func (c *commandS3LifecycleRunShard) HasTag(CommandTag) bool { return false }
+
+func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer io.Writer) error {
+	fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	shard := fs.Int("shard", -1, "single shard id in [0, 16); use -shards for a range or set")
+	shardsSpec := fs.String("shards", "", "shard range \"lo-hi\" or comma list \"a,b,c\"; mutually exclusive with -shard")
+	s3Endpoint := fs.String("s3", "", "s3 server gRPC endpoint, host:port")
+	eventBudget := fs.Int("events", 1000, "max in-shard events to process before returning (0 = unbounded; counts only events that pass the shard filter)")
+	dispatchTick := fs.Duration("dispatch", 5*time.Second, "dispatcher tick cadence")
+	checkpointTick := fs.Duration("checkpoint", 30*time.Second, "cursor checkpoint cadence")
+	runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout. -events alone can hang on quiet shards")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	shards, err := resolveShardSelection(*shard, *shardsSpec)
+	if err != nil {
+		return err
+	}
+	if *s3Endpoint == "" {
+		return fmt.Errorf("-s3 required (host:port of s3 server gRPC)")
+	}
+	if *eventBudget < 0 {
+		return fmt.Errorf("-events must be >= 0 (0 = unbounded)")
+	}
+
+	bucketsPath, err := resolveBucketsPath(env)
+	if err != nil {
+		return fmt.Errorf("resolve buckets path: %w", err)
+	}
+	fmt.Fprintf(writer, "buckets path: %s\n", bucketsPath)
+
+	dialCtx, dialCancel := context.WithTimeout(context.Background(), 30*time.Second)
+	conn, err := pb.GrpcDial(dialCtx, *s3Endpoint, false, env.option.GrpcDialOption)
+	dialCancel()
+	if err != nil {
+		return fmt.Errorf("dial s3 %s: %w", *s3Endpoint, err)
+	}
+	defer conn.Close()
+	rpcClient := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(conn)
+
+	// Run the whole pipeline inside one WithFilerClient so the reader's
+	// SubscribeMetadata stream and the persister share a single connection.
+	return env.WithFilerClient(true, func(filerClient filer_pb.SeaweedFilerClient) error {
+		inputs, parseErrors, err := loadLifecycleCompileInputs(context.Background(), filerClient, bucketsPath)
+		if err != nil {
+			return fmt.Errorf("load lifecycle configs: %w", err)
+		}
+		for i, pe := range parseErrors {
+			// Surface up to the first three parse errors so the operator
+			// can chase malformed configs; cap the rest with a count so
+			// the output stays readable on large clusters.
+			if i < 3 {
+				fmt.Fprintf(writer, "warning: %s: %v\n", pe.bucket, pe.err)
+			}
+		}
+		if extra := len(parseErrors) - 3; extra > 0 {
+			fmt.Fprintf(writer, "warning: %d additional bucket(s) had malformed lifecycle config\n", extra)
+		}
+		if len(inputs) == 0 {
+			fmt.Fprintln(writer, "no buckets with enabled lifecycle rules found")
+			return nil
+		}
+		fmt.Fprintf(writer, "loaded lifecycle for %d bucket(s)\n", len(inputs))
+
+		// Activate every action so this manual run dispatches whatever fires.
+		// The production bootstrap walker promotes actions only after a clean
+		// walk; this shell entrypoint runs out-of-band of that flow.
+		eng := engine.New()
+		eng.Compile(inputs, engine.CompileOptions{PriorStates: allActivePriorStates(inputs)})
+
+		pipeline := &dispatcher.Pipeline{
+			Shards:         shards,
+			BucketsPath:    bucketsPath,
+			Engine:         eng,
+			Persister:      &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
+			Client:         &lifecycleClientCallable{c: rpcClient},
+			FilerClient:    filerClient,
+			ClientID:       util.RandomInt32(),
+			ClientName:     fmt.Sprintf("shell-lifecycle-%s", formatShardLabel(shards)),
+			DispatchTick:   *dispatchTick,
+			CheckpointTick: *checkpointTick,
+			EventBudget:    *eventBudget,
+		}
+
+		var ctx context.Context
+		var cancel context.CancelFunc
+		if *runtime > 0 {
+			ctx, cancel = context.WithTimeout(context.Background(), *runtime)
+		} else {
+			ctx, cancel = context.WithCancel(context.Background())
+		}
+		defer cancel()
+
+		fmt.Fprintf(writer, "running shards %s (event budget=%d, runtime=%s)…\n", formatShardLabel(shards), *eventBudget, *runtime)
+		if err := pipeline.Run(ctx); err != nil {
+			return fmt.Errorf("pipeline: %w", err)
+		}
+		fmt.Fprintf(writer, "shards %s complete; cursors checkpointed\n", formatShardLabel(shards))
+		return nil
+	})
+}
+
+// resolveShardSelection turns the -shard / -shards flags into a sorted,
+// deduplicated []int. Exactly one form must be specified.
+func resolveShardSelection(singleShard int, shardsSpec string) ([]int, error) {
+	if singleShard >= 0 && shardsSpec != "" {
+		return nil, fmt.Errorf("-shard and -shards are mutually exclusive")
+	}
+	if singleShard < 0 && shardsSpec == "" {
+		return nil, fmt.Errorf("specify -shard <id> or -shards <range|set>")
+	}
+	if singleShard >= 0 {
+		if singleShard >= s3lifecycle.ShardCount {
+			return nil, fmt.Errorf("-shard %d out of [0,%d)", singleShard, s3lifecycle.ShardCount)
+		}
+		return []int{singleShard}, nil
+	}
+	return parseShardsSpec(shardsSpec)
+}
+
+// parseShardsSpec accepts "lo-hi" (inclusive) or "a,b,c" and returns a
+// sorted, deduplicated, in-range []int.
+func parseShardsSpec(spec string) ([]int, error) {
+	spec = strings.TrimSpace(spec)
+	seen := map[int]struct{}{}
+	add := func(v int) error {
+		if v < 0 || v >= s3lifecycle.ShardCount {
+			return fmt.Errorf("shard %d out of [0,%d)", v, s3lifecycle.ShardCount)
+		}
+		seen[v] = struct{}{}
+		return nil
+	}
+	if strings.Contains(spec, "-") && !strings.Contains(spec, ",") {
+		parts := strings.SplitN(spec, "-", 2)
+		lo, err := strconv.Atoi(strings.TrimSpace(parts[0]))
+		if err != nil {
+			return nil, fmt.Errorf("range lo: %w", err)
+		}
+		hi, err := strconv.Atoi(strings.TrimSpace(parts[1]))
+		if err != nil {
+			return nil, fmt.Errorf("range hi: %w", err)
+		}
+		if lo > hi {
+			return nil, fmt.Errorf("range lo %d > hi %d", lo, hi)
+		}
+		for v := lo; v <= hi; v++ {
+			if err := add(v); err != nil {
+				return nil, err
+			}
+		}
+	} else {
+		for _, part := range strings.Split(spec, ",") {
+			part = strings.TrimSpace(part)
+			if part == "" {
+				continue
+			}
+			v, err := strconv.Atoi(part)
+			if err != nil {
+				return nil, fmt.Errorf("shard list: %w", err)
+			}
+			if err := add(v); err != nil {
+				return nil, err
+			}
+		}
+	}
+	if len(seen) == 0 {
+		return nil, fmt.Errorf("empty shard set")
+	}
+	out := make([]int, 0, len(seen))
+	for v := range seen {
+		out = append(out, v)
+	}
+	sort.Ints(out)
+	return out, nil
+}
+
+func formatShardLabel(shards []int) string {
+	if len(shards) == 1 {
+		return fmt.Sprintf("%d", shards[0])
+	}
+	// Detect contiguous range.
+	contiguous := true
+	for i := 1; i < len(shards); i++ {
+		if shards[i] != shards[i-1]+1 {
+			contiguous = false
+			break
+		}
+	}
+	if contiguous {
+		return fmt.Sprintf("%d-%d", shards[0], shards[len(shards)-1])
+	}
+	parts := make([]string, len(shards))
+	for i, v := range shards {
+		parts[i] = strconv.Itoa(v)
+	}
+	return strings.Join(parts, ",")
+}
+
+// lifecycleClientCallable adapts the generated grpc client (variadic
+// CallOption tail) to the dispatcher.LifecycleClient interface.
+type lifecycleClientCallable struct {
+	c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient
+}
+
+func (l *lifecycleClientCallable) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+	return l.c.LifecycleDelete(ctx, req)
+}
+
+// resolveBucketsPath fetches the filer's configured buckets directory.
+// Falls back to /buckets when the filer doesn't return one.
+func resolveBucketsPath(env *CommandEnv) (string, error) {
+	var path string
+	err := env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		path = resp.GetDirBuckets()
+		return nil
+	})
+	if err != nil {
+		return "", err
+	}
+	if path == "" {
+		path = "/buckets"
+	}
+	return path, nil
+}
+
+type lifecycleParseError struct {
+	bucket string
+	err    error
+}
+
+// loadLifecycleCompileInputs walks the buckets directory and reads each
+// bucket entry's lifecycle XML from its Extended attributes. Pagination
+// loops with startFrom so clusters with more than one page of buckets
+// don't drop the tail. Parse errors are collected per bucket and returned
+// alongside the successful inputs so the caller can surface them.
+func loadLifecycleCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketsPath string) ([]engine.CompileInput, []lifecycleParseError, error) {
+	var (
+		inputs      []engine.CompileInput
+		parseErrors []lifecycleParseError
+		startFrom   string
+	)
+	const pageSize uint32 = 1024
+	for {
+		pageCount := 0
+		var lastName string
+		err := filer_pb.SeaweedList(ctx, client, bucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+			pageCount++
+			lastName = entry.Name
+			if !entry.IsDirectory {
+				return nil
+			}
+			xmlBytes, ok := entry.Extended[bucketLifecycleConfigurationXMLKey]
+			if !ok || len(xmlBytes) == 0 {
+				return nil
+			}
+			rules, err := lifecycle_xml.ParseCanonical(xmlBytes)
+			if err != nil {
+				parseErrors = append(parseErrors, lifecycleParseError{bucket: entry.Name, err: err})
+				return nil
+			}
+			if len(rules) == 0 {
+				return nil
+			}
+			inputs = append(inputs, engine.CompileInput{
+				Bucket:    entry.Name,
+				Rules:     rules,
+				Versioned: isBucketVersioned(entry),
+			})
+			return nil
+		}, startFrom, false, pageSize)
+		if err != nil {
+			return nil, nil, err
+		}
+		if uint32(pageCount) < pageSize {
+			break
+		}
+		startFrom = lastName
+	}
+	return inputs, parseErrors, nil
+}
+
+const bucketLifecycleConfigurationXMLKey = "s3-bucket-lifecycle-configuration-xml"
+
+func isBucketVersioned(entry *filer_pb.Entry) bool {
+	v, ok := entry.Extended[s3_constants.ExtVersioningKey]
+	if !ok {
+		return false
+	}
+	s := strings.ToLower(strings.TrimSpace(string(v)))
+	return s == "enabled" || s == "suspended"
+}
+
+// allActivePriorStates seeds every compiled action as bootstrap-complete +
+// event-driven so the run dispatches whatever fires. Production bootstrap
+// walks set this incrementally per bucket; this manual run skips the walk.
+func allActivePriorStates(inputs []engine.CompileInput) map[s3lifecycle.ActionKey]engine.PriorState {
+	prior := map[s3lifecycle.ActionKey]engine.PriorState{}
+	for _, in := range inputs {
+		for _, rule := range in.Rules {
+			hash := s3lifecycle.RuleHash(rule)
+			for _, kind := range s3lifecycle.RuleActionKinds(rule) {
+				key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: hash, ActionKind: kind}
+				prior[key] = engine.PriorState{
+					BootstrapComplete: true,
+					Mode:              engine.ModeEventDriven,
+				}
+			}
+		}
+	}
+	return prior
+}

weed/shell/command_s3_lifecycle_run_shard_test.go

@@ -0,0 +1,92 @@
+package shell
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestParseShardsSpec_Range(t *testing.T) {
+	got, err := parseShardsSpec("3-7")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	want := []int{3, 4, 5, 6, 7}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestParseShardsSpec_Set(t *testing.T) {
+	got, err := parseShardsSpec("0,3,7")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	want := []int{0, 3, 7}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestParseShardsSpec_DedupSort(t *testing.T) {
+	got, err := parseShardsSpec("7,3,3,0")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	want := []int{0, 3, 7}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestParseShardsSpec_OutOfRange(t *testing.T) {
+	if _, err := parseShardsSpec("16"); err == nil {
+		t.Fatal("expected out-of-range error for 16")
+	}
+	if _, err := parseShardsSpec("-1"); err == nil {
+		t.Fatal("expected out-of-range error for -1")
+	}
+	if _, err := parseShardsSpec("0-16"); err == nil {
+		t.Fatal("expected out-of-range error for 0-16")
+	}
+}
+
+func TestParseShardsSpec_BadRange(t *testing.T) {
+	if _, err := parseShardsSpec("7-3"); err == nil {
+		t.Fatal("expected lo>hi error")
+	}
+}
+
+func TestResolveShardSelection_Mutex(t *testing.T) {
+	if _, err := resolveShardSelection(0, "1,2"); err == nil {
+		t.Fatal("expected mutex error when both -shard and -shards set")
+	}
+	if _, err := resolveShardSelection(-1, ""); err == nil {
+		t.Fatal("expected error when neither set")
+	}
+}
+
+func TestResolveShardSelection_SingleShard(t *testing.T) {
+	got, err := resolveShardSelection(5, "")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if !reflect.DeepEqual(got, []int{5}) {
+		t.Fatalf("got %v, want [5]", got)
+	}
+}
+
+func TestFormatShardLabel(t *testing.T) {
+	cases := []struct {
+		in   []int
+		want string
+	}{
+		{[]int{5}, "5"},
+		{[]int{0, 1, 2, 3}, "0-3"},
+		{[]int{0, 2, 5}, "0,2,5"},
+	}
+	for _, tc := range cases {
+		if got := formatShardLabel(tc.in); got != tc.want {
+			t.Errorf("formatShardLabel(%v)=%q, want %q", tc.in, got, tc.want)
+		}
+	}
+}