Merge remote-tracking branch 'origin/master' into fix-delete-offline-replica

# Conflicts:
#	weed/storage/store_mark_readonly_can_delete_test.go

.github/workflows/rust-vector-tests.yml

@@ -76,19 +76,25 @@ jobs:
 
 
   vectors-e2e:
-    name: S3 Vectors Integration (${{ matrix.volume }} volume)
+    name: S3 Vectors Integration (${{ matrix.volume }} volume${{ matrix.hardened && ', hardened' || '' }})
     runs-on: ubuntu-24.04
     timeout-minutes: 45
     strategy:
       fail-fast: false
       matrix:
         volume: [go, rust]
+        # Hardened: volume servers behind mutual TLS, read JWTs required,
+        # and every chunk encrypted by the gateway.
+        hardened: ["", "1"]
     env:
       CARGO_PROFILE_DEV_DEBUG: "0"
       CARGO_BUILD_JOBS: "2"
       CARGO_TARGET_DIR: ${{ github.workspace }}/seaweed-vector/target
       VOLUME_SERVER_IMPL: ${{ matrix.volume }}
       VECTOR_QUERY_MODE: single
+      VECTOR_TLS: ${{ matrix.hardened }}
+      VECTOR_READ_JWT: ${{ matrix.hardened }}
+      VECTOR_ENCRYPT_VOLUMES: ${{ matrix.hardened }}
     steps:
       - uses: actions/checkout@v7
         with:
@@ -147,7 +153,7 @@ jobs:
         if: always()
         uses: actions/upload-artifact@v7
         with:
-          name: vectors-e2e-diagnostics-${{ matrix.volume }}
+          name: vectors-e2e-diagnostics-${{ matrix.volume }}${{ matrix.hardened && '-hardened' || '' }}
           path: |
             ${{ runner.temp }}/vector-tests
             ${{ runner.temp }}/vector-results.xml

test/s3vectors/conftest.py

@@ -1,11 +1,20 @@
+import base64
 import contextlib
+import gzip
+import hashlib
+import hmac
+import http.client
+import json
 import os
 from pathlib import Path
 import socket
+import ssl
 import subprocess
 import time
 import urllib.error
 import urllib.request
+import uuid
+import zlib
 
 import boto3
 from botocore.config import Config
@@ -79,6 +88,85 @@ def open_cluster(request, tmp_path_factory):
     yield from run_cluster(request.param, tmp_path_factory, identities=False)
 
 
+def write_tls(directory):
+    """A CA and one certificate for localhost; volume servers serve HTTPS
+    requiring client certificates, and every client presents the same one."""
+    def openssl(*args):
+        subprocess.run(["openssl", *args], check=True, capture_output=True, cwd=directory)
+    openssl("req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "ca.key", "-out", "ca.pem",
+            "-subj", "/CN=s3vectors test CA", "-days", "2",
+            "-addext", "basicConstraints=critical,CA:TRUE", "-addext", "keyUsage=critical,keyCertSign,cRLSign")
+    openssl("req", "-newkey", "rsa:2048", "-nodes", "-keyout", "key.pem", "-out", "req.csr", "-subj", "/CN=localhost")
+    (directory / "ext.cnf").write_text(
+        "subjectAltName=DNS:localhost,IP:127.0.0.1\nextendedKeyUsage=serverAuth,clientAuth\nbasicConstraints=CA:FALSE\n")
+    openssl("x509", "-req", "-in", "req.csr", "-CA", "ca.pem", "-CAkey", "ca.key", "-CAcreateserial",
+            "-out", "cert.pem", "-days", "2", "-extfile", "ext.cnf")
+    pem = "".join(f'{key} = {json.dumps(str(directory / name), ensure_ascii=False)}\n' for key, name in (("cert", "cert.pem"), ("key", "key.pem"), ("ca", "ca.pem")))
+    (directory / "security.toml").write_text(f"[https.volume]\n{pem}\n[https.client]\nenabled = true\n{pem}")
+
+
+def read_token(key, fid):
+    def encode(data):
+        return base64.urlsafe_b64encode(data).rstrip(b"=")
+    signed = encode(b'{"alg":"HS256","typ":"JWT"}') + b"." + encode(json.dumps({"fid": fid}).encode())
+    return (signed + b"." + encode(hmac.new(key.encode(), signed, hashlib.sha256).digest())).decode()
+
+
+def probe_hardening(endpoint, filer, vol, directory, read_jwt_key):
+    """The hardened switches must be enforced, not only configured: a TLS
+    volume server refuses plain HTTP and a connection without the client
+    certificate, a volume server under a read key refuses an unsigned
+    read of a real chunk, and an encrypted chunk is stored as ciphertext."""
+    scheme = "https" if os.environ.get("VECTOR_TLS") else "http"
+    context = None
+    if scheme == "https":
+        context = ssl.create_default_context(cafile=directory / "ca.pem")
+        for url, kwargs in ((f"http://127.0.0.1:{vol}/status", {}),
+                            (f"https://127.0.0.1:{vol}/status", {"context": context})):
+            try:
+                with urllib.request.urlopen(url, timeout=5, **kwargs) as response:
+                    assert response.status != 200, f"{url} answered without the client certificate"
+            # Go answers plaintext with HTTP 400, Rust with a TLS alert.
+            except (urllib.error.URLError, ConnectionError, ssl.SSLError, http.client.HTTPException):
+                pass
+        context.load_cert_chain(directory / "cert.pem", directory / "key.pem")
+        with urllib.request.urlopen(f"https://127.0.0.1:{vol}/status", timeout=5, context=context) as response:
+            assert response.status == 200
+    encrypted = os.environ.get("VECTOR_ENCRYPT_VOLUMES")
+    if not (read_jwt_key or encrypted):
+        return
+    s3 = boto3.client("s3", endpoint_url=endpoint, aws_access_key_id="test", aws_secret_access_key="test",
+                      region_name="us-east-1", config=Config(s3={"addressing_style": "path"}))
+    bucket = "hardening-" + uuid.uuid4().hex[:12]
+    body = b"x" * 1024
+    s3.create_bucket(Bucket=bucket)
+    try:
+        s3.put_object(Bucket=bucket, Key="chunk", Body=body)
+        with urllib.request.urlopen(f"http://127.0.0.1:{filer}/buckets/{bucket}/chunk?metadata=true", timeout=5) as response:
+            chunk = json.load(response)["chunks"][0]
+        url = f"{scheme}://127.0.0.1:{vol}/{chunk['file_id']}"
+        headers = {}
+        if read_jwt_key:
+            try:
+                with urllib.request.urlopen(url, timeout=5, context=context):
+                    raise AssertionError("the volume server served an unsigned read")
+            except urllib.error.HTTPError as error:
+                assert error.code == 401, error
+            headers["Authorization"] = "Bearer " + read_token(read_jwt_key, chunk["file_id"])
+        with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=5, context=context) as response:
+            stored = response.read()
+        if encrypted:
+            assert chunk.get("cipher_key"), "the gateway stored a chunk without a cipher key"
+            assert stored and stored != body, "the volume server holds the chunk in plaintext or not at all"
+            with contextlib.suppress(OSError, EOFError, zlib.error):
+                assert gzip.decompress(stored) != body, "the volume server holds the chunk gzipped, not encrypted"
+        else:
+            assert stored == body, "the signed read did not return the chunk"
+    finally:
+        s3.delete_object(Bucket=bucket, Key="chunk")
+        s3.delete_bucket(Bucket=bucket)
+
+
 def run_cluster(param, tmp_path_factory, identities):
     volume, mode = param
     directory = tmp_path_factory.mktemp(f"vectors-{volume}-{mode}" + ("" if identities else "-open"))
@@ -99,12 +187,23 @@ def run_cluster(param, tmp_path_factory, identities):
         master_addr = f"127.0.0.1:{master}.{master_grpc}"
         filer_addr = f"127.0.0.1:{filer}.{filer_grpc}"
 
+        # VECTOR_READ_JWT=1 makes volume reads need a JWT; every weed process
+        # and the workers sign with the key.
+        read_jwt_key = os.environ.get("VECTOR_READ_JWT") and uuid.uuid4().hex
+        # VECTOR_TLS=1 puts the volume servers behind HTTPS with client
+        # certificates; every process finds security.toml in its cwd.
+        if os.environ.get("VECTOR_TLS"):
+            write_tls(directory)
+
         def start(name, args, unset=()):
             log = stack.enter_context((directory / f"{name}.log").open("w"))
             env = dict(os.environ, AWS_ACCESS_KEY_ID="test", AWS_SECRET_ACCESS_KEY="test",
                        AWS_DEFAULT_REGION="us-east-1", AWS_EC2_METADATA_DISABLED="true",
-                       WEED_S3_SSE_KEY="s3vectors-test")
+                       WEED_S3_SSE_KEY="s3vectors-test",
+                       **({"WEED_JWT_SIGNING_READ_KEY": read_jwt_key} if read_jwt_key else {}))
             env.pop("WEED_S3_SSE_KEK", None)
+            if not read_jwt_key:
+                env.pop("WEED_JWT_SIGNING_READ_KEY", None)
             for key in unset:
                 env.pop(key)
             process = subprocess.Popen(args, cwd=directory, stdout=log, stderr=log, env=env)
@@ -139,10 +238,15 @@ def run_cluster(param, tmp_path_factory, identities):
                            "--storage-option", "aws_secret_access_key=test"]
         else:
             config, unset, credentials = [], ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"), []
-        start("s3", [weed, "-v=1", "s3", f"-filer={filer_addr}", *config,
+        # VECTOR_ENCRYPT_VOLUMES=1 has the gateway encrypt every chunk it
+        # stores, which direct reads must undo.
+        encrypt = ["-encryptVolumeData"] if os.environ.get("VECTOR_ENCRYPT_VOLUMES") else []
+        start("s3", [weed, "-v=1", "s3", f"-filer={filer_addr}", *config, *encrypt,
                      "-ip=127.0.0.1", "-ip.bind=127.0.0.1", f"-port={s3}",
                      f"-port.grpc={s3_grpc}", "-port.iceberg=0", f"-port.lance={lance}"], unset)
         wait_for(lambda: http_ready(f"http://127.0.0.1:{lance}/v1/namespace/root/list"), processes, "catalog not ready")
+        if os.environ.get("VECTOR_TLS") or read_jwt_key or encrypt:
+            probe_hardening(endpoint, filer, vol, directory, read_jwt_key)
         start("vector", [engine, "--namespace", f"http://127.0.0.1:{lance}",
                          "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", *credentials], unset)
         wait_for(lambda: listening(vector), processes, "vector engine not ready")

weed/admin/dash/admin_server.go

@@ -2309,17 +2309,25 @@ func collectCollectionStats(topologyInfo *master_pb.TopologyInfo) map[string]col
 
 	// Fold EC per-volume counts into the collection totals. fileCount is
 	// deduped via max across every node reporting shards for the volume;
-	// deleteCount is summed across the same nodes.
+	// deleteCount is summed across the same nodes, and a tombstone recorded
+	// on more than one shard holder can push that sum past fileCount. This
+	// is an expected accounting artefact that recurs on every refresh, so
+	// it is summarized once per pass rather than logged per volume.
+	skipped := 0
 	for vid, agg := range ecVolumeAgg {
 		data := collectionMap[agg.collection]
 		if agg.fileCount >= agg.deleteCount {
 			data.FileCount += int64(agg.fileCount - agg.deleteCount)
 		} else {
-			glog.Warningf("ec volume %d in collection %q: summed delete_count=%d exceeds file_count=%d; skipping object count",
+			skipped++
+			glog.V(2).Infof("ec volume %d in collection %q: summed delete_count=%d exceeds file_count=%d; skipping object count",
 				vid, agg.collection, agg.deleteCount, agg.fileCount)
 		}
 		collectionMap[agg.collection] = data
 	}
+	if skipped > 0 {
+		glog.V(1).Infof("%d ec volumes have summed delete_count above file_count; skipping object count for them", skipped)
+	}
 
 	return collectionMap
 }

weed/admin/dash/collect_collection_stats_test.go

@@ -185,6 +185,47 @@ func TestCollectCollectionStatsECFileAndDeleteCountAggregation(t *testing.T) {
 	}
 }
 
+// TestCollectCollectionStatsECDeleteCountAboveFileCount verifies that an EC
+// volume whose summed delete_count exceeds its file_count adds nothing to the
+// collection's object count, while other volumes in the collection still do.
+func TestCollectCollectionStatsECDeleteCountAboveFileCount(t *testing.T) {
+	shard := func(vid uint32, bits uint32, fileCount, deleteCount uint64) *master_pb.VolumeEcShardInformationMessage {
+		return &master_pb.VolumeEcShardInformationMessage{
+			Id:          vid,
+			Collection:  "bucket-c",
+			EcIndexBits: bits,
+			ShardSizes:  []int64{1},
+			FileCount:   fileCount,
+			DeleteCount: deleteCount,
+		}
+	}
+	node := func(shards ...*master_pb.VolumeEcShardInformationMessage) *master_pb.DataNodeInfo {
+		return &master_pb.DataNodeInfo{
+			DiskInfos: map[string]*master_pb.DiskInfo{"disk1": {EcShardInfos: shards}},
+		}
+	}
+
+	topo := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{
+			{
+				RackInfos: []*master_pb.RackInfo{
+					{
+						DataNodeInfos: []*master_pb.DataNodeInfo{
+							node(shard(21, 1<<0, 100, 60), shard(22, 1<<0, 50, 10)),
+							node(shard(21, 1<<1, 100, 60)),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	got := collectCollectionStats(topo)["bucket-c"]
+	if got.FileCount != 40 {
+		t.Errorf("FileCount: got %d, want 40 (volume 21 skipped, volume 22 has 50-10)", got.FileCount)
+	}
+}
+
 // TestCollectCollectionStatsECFileCountMaxDedupe verifies that EC file_count
 // is taken as the max across reporting nodes rather than the first-seen
 // value. A node that has not yet finished loading .ecx reports file_count=0,

weed/admin/dash/plugin_api.go

@@ -152,12 +152,21 @@ func (s *AdminServer) GetPluginJobsAPI(w http.ResponseWriter, r *http.Request) {
 	state := strings.TrimSpace(query.Get("state"))
 	laneFilter := strings.TrimSpace(query.Get("lane"))
 	limit := parsePositiveInt(query.Get("limit"), 200)
-	jobs := s.ListPluginJobs(jobType, state, limit)
+	// Apply the lane before the limit, or a busy lane's newest rows crowd out a quiet lane.
+	listLimit := limit
+	if laneFilter != "" {
+		listLimit = 0
+	}
+	jobs := s.ListPluginJobs(jobType, state, listLimit)
 	if jobs == nil {
 		writeJSON(w, http.StatusOK, []interface{}{})
 		return
 	}
-	writeJSON(w, http.StatusOK, filterTrackedJobsByLane(jobs, laneFilter))
+	jobs = filterTrackedJobsByLane(jobs, laneFilter)
+	if len(jobs) > limit {
+		jobs = jobs[:limit]
+	}
+	writeJSON(w, http.StatusOK, jobs)
 }
 
 // GetPluginJobAPI returns one tracked job.
@@ -248,12 +257,20 @@ func (s *AdminServer) GetPluginActivitiesAPI(w http.ResponseWriter, r *http.Requ
 	jobType := strings.TrimSpace(query.Get("job_type"))
 	laneFilter := strings.TrimSpace(query.Get("lane"))
 	limit := parsePositiveInt(query.Get("limit"), 500)
-	activities := s.ListPluginActivities(jobType, limit)
+	listLimit := limit
+	if laneFilter != "" {
+		listLimit = 0
+	}
+	activities := s.ListPluginActivities(jobType, listLimit)
 	if activities == nil {
 		writeJSON(w, http.StatusOK, []interface{}{})
 		return
 	}
-	writeJSON(w, http.StatusOK, filterActivitiesByLane(activities, laneFilter))
+	activities = filterActivitiesByLane(activities, laneFilter)
+	if len(activities) > limit {
+		activities = activities[:limit]
+	}
+	writeJSON(w, http.StatusOK, activities)
 }
 
 // GetPluginSchedulerStatesAPI returns per-job-type scheduler status for monitoring.

weed/admin/dash/plugin_api_test.go

@@ -3,11 +3,13 @@ package dash
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/seaweedfs/seaweedfs/weed/admin/plugin"
@@ -230,3 +232,61 @@ func TestFilterActivitiesByLane(t *testing.T) {
 		t.Fatalf("expected lifecycle activity to be retained, got %+v", filtered[0])
 	}
 }
+
+func TestPluginJobsAndActivitiesAPILimitAfterLaneFilter(t *testing.T) {
+	dataDir := t.TempDir()
+	store, err := plugin.NewConfigStore(dataDir)
+	if err != nil {
+		t.Fatalf("NewConfigStore: %v", err)
+	}
+	now := time.Now()
+	var jobs []plugin.TrackedJob
+	var activities []plugin.JobActivity
+	for i := 0; i < 5; i++ {
+		jobType, jobID := "vacuum", fmt.Sprintf("vacuum-%d", i)
+		if i == 4 {
+			jobType, jobID = "iceberg_maintenance", "iceberg-1"
+		}
+		at := now.Add(-time.Duration(i) * time.Minute)
+		jobs = append(jobs, plugin.TrackedJob{JobID: jobID, JobType: jobType, State: plugin.StateSucceeded, UpdatedAt: &at})
+		activities = append(activities, plugin.JobActivity{JobID: jobID, JobType: jobType, Message: "done", OccurredAt: &at})
+	}
+	if err := store.SaveTrackedJobs(jobs); err != nil {
+		t.Fatalf("SaveTrackedJobs: %v", err)
+	}
+	if err := store.SaveActivities(activities); err != nil {
+		t.Fatalf("SaveActivities: %v", err)
+	}
+	p, err := plugin.New(plugin.Options{DataDir: dataDir})
+	if err != nil {
+		t.Fatalf("plugin.New: %v", err)
+	}
+	defer p.Shutdown()
+	s := &AdminServer{plugin: p}
+
+	get := func(handler http.HandlerFunc, url string) []map[string]any {
+		recorder := httptest.NewRecorder()
+		handler(recorder, httptest.NewRequest(http.MethodGet, url, nil))
+		if recorder.Code != http.StatusOK {
+			t.Fatalf("%s: status %d", url, recorder.Code)
+		}
+		var rows []map[string]any
+		if err := json.Unmarshal(recorder.Body.Bytes(), &rows); err != nil {
+			t.Fatalf("%s: %v", url, err)
+		}
+		return rows
+	}
+
+	if rows := get(s.GetPluginJobsAPI, "/api/plugin/jobs?limit=2&lane=iceberg"); len(rows) != 1 || rows[0]["job_id"] != "iceberg-1" {
+		t.Fatalf("iceberg lane jobs: got %v, want the one iceberg job", rows)
+	}
+	if rows := get(s.GetPluginJobsAPI, "/api/plugin/jobs?limit=2&lane=default"); len(rows) != 2 {
+		t.Fatalf("default lane jobs: got %d rows, want the limit of 2", len(rows))
+	}
+	if rows := get(s.GetPluginActivitiesAPI, "/api/plugin/activities?limit=2&lane=iceberg"); len(rows) != 1 || rows[0]["job_id"] != "iceberg-1" {
+		t.Fatalf("iceberg lane activities: got %v, want the one iceberg activity", rows)
+	}
+	if rows := get(s.GetPluginActivitiesAPI, "/api/plugin/activities?limit=2&lane=default"); len(rows) != 2 {
+		t.Fatalf("default lane activities: got %d rows, want the limit of 2", len(rows))
+	}
+}

weed/admin/view/app/plugin.templ

@@ -620,6 +620,8 @@ templ Plugin(page string, initialJob string, lane string) {
                 schedulerLock: null,
                 allJobs: [],
                 allActivities: [],
+                monitorRowsScope: null,
+                refreshError: '',
                 loadedJobType: '',
                 descriptor: null,
                 config: null,
@@ -633,6 +635,13 @@ templ Plugin(page string, initialJob string, lane string) {
                 return url + sep + 'lane=' + encodeURIComponent(pluginLane);
             }
 
+            // appendJobTypeParam appends ?job_type= or &job_type= to a URL if a job type is given.
+            function appendJobTypeParam(url, jobType) {
+                if (!jobType) return url;
+                var sep = url.indexOf('?') === -1 ? '?' : '&';
+                return url + sep + 'job_type=' + encodeURIComponent(jobType);
+            }
+
             function getCSRFToken() {
                 var meta = document.querySelector('meta[name="csrf-token"]');
                 if (!meta) {
@@ -787,6 +796,10 @@ templ Plugin(page string, initialJob string, lane string) {
                 return { jobType: jobType, displayName: displayName };
             }
 
+            function monitorJobTypeScope() {
+                return state.activeTopTab === 'overview' ? '' : state.selectedJobType;
+            }
+
             function topTabKeyForJobType(jobType) {
                 var normalized = String(jobType || '').trim();
                 if (!normalized) {
@@ -1257,11 +1270,16 @@ templ Plugin(page string, initialJob string, lane string) {
                     } else {
                         notify('Job expired: ' + normalizedJobID, 'success');
                     }
-                    await refreshJobsAndActivities();
-                    await openJobDetail(normalizedJobID);
                 } catch (e) {
                     notify('Failed to expire job: ' + e.message, 'error');
+                    return;
                 }
+                try {
+                    await refreshJobsAndActivities();
+                } catch (refreshErr) {
+                    notify('Monitoring refresh failed: ' + refreshErr.message, 'error');
+                }
+                await openJobDetail(normalizedJobID);
             }
 
             async function openJobDetail(jobID) {
@@ -1608,7 +1626,7 @@ templ Plugin(page string, initialJob string, lane string) {
             }
 
             function renderSchedulerStatus() {
-                window.renderPluginSchedulerStatus({scheduler: state.schedulerStatus, lock: state.schedulerLock, job_type: state.activeTopTab === 'overview' ? '' : state.selectedJobType});
+                window.renderPluginSchedulerStatus({scheduler: state.schedulerStatus, lock: state.schedulerLock, job_type: monitorJobTypeScope()});
                 var valueNodes = document.querySelectorAll('.plugin-scheduler-next-run');
                 if (!valueNodes.length) {
                     return;
@@ -1735,9 +1753,9 @@ templ Plugin(page string, initialJob string, lane string) {
                 }
 
                 var filteredJobs = [];
-                var allJobs = Array.isArray(state.allJobs) ? state.allJobs : [];
-                for (var i = 0; i < allJobs.length; i++) {
-                    var job = allJobs[i] || {};
+                var jobs = Array.isArray(state.jobs) ? state.jobs : [];
+                for (var i = 0; i < jobs.length; i++) {
+                    var job = jobs[i] || {};
                     var jobType = String(job.job_type || '').trim();
                     if (state.selectedJobType && jobType !== state.selectedJobType) {
                         continue;
@@ -1801,9 +1819,9 @@ templ Plugin(page string, initialJob string, lane string) {
                 }
 
                 var filteredJobs = [];
-                var allJobs = Array.isArray(state.allJobs) ? state.allJobs : [];
-                for (var i = 0; i < allJobs.length; i++) {
-                    var queueJob = allJobs[i] || {};
+                var jobs = Array.isArray(state.jobs) ? state.jobs : [];
+                for (var i = 0; i < jobs.length; i++) {
+                    var queueJob = jobs[i] || {};
                     var queueJobType = String(queueJob.job_type || '').trim();
                     if (state.selectedJobType && queueJobType !== state.selectedJobType) {
                         continue;
@@ -1858,9 +1876,9 @@ templ Plugin(page string, initialJob string, lane string) {
                 }
 
                 var filteredActivities = [];
-                var allActivities = Array.isArray(state.allActivities) ? state.allActivities : [];
-                for (var i = 0; i < allActivities.length; i++) {
-                    var activity = allActivities[i] || {};
+                var activities = Array.isArray(state.activities) ? state.activities : [];
+                for (var i = 0; i < activities.length; i++) {
+                    var activity = activities[i] || {};
                     var jobType = String(activity.job_type || '').trim();
                     if (state.selectedJobType && jobType !== state.selectedJobType) {
                         continue;
@@ -1900,9 +1918,9 @@ templ Plugin(page string, initialJob string, lane string) {
                 }
 
                 var filteredActivities = [];
-                var allActivities = Array.isArray(state.allActivities) ? state.allActivities : [];
-                for (var i = 0; i < allActivities.length; i++) {
-                    var activity = allActivities[i] || {};
+                var activities = Array.isArray(state.activities) ? state.activities : [];
+                for (var i = 0; i < activities.length; i++) {
+                    var activity = activities[i] || {};
                     if (isDetectionActivitySource(activity.source)) {
                         continue;
                     }
@@ -3064,9 +3082,14 @@ templ Plugin(page string, initialJob string, lane string) {
                         '<div class="text-success">Success: ' + escapeHtml(String(data.success_count || 0)) + '</div>' +
                         '<div class="text-danger">Errors: ' + escapeHtml(String(data.error_count || 0)) + '</div>';
                     await loadRunHistory(jobType);
-                    await refreshJobsAndActivities();
                 } catch (e) {
                     notify('Workflow run failed: ' + e.message, 'error');
+                    return;
+                }
+                try {
+                    await refreshJobsAndActivities();
+                } catch (refreshErr) {
+                    notify('Workflow run succeeded, but monitoring refresh failed: ' + refreshErr.message, 'error');
                 }
             }
 
@@ -3075,13 +3098,31 @@ templ Plugin(page string, initialJob string, lane string) {
 
                 state.monitorExecutionJobState = executionStateFilter ? String(executionStateFilter.value || '').trim() : '';
 
-                var allJobsPromise = pluginRequest('GET', appendLaneParam('/api/plugin/jobs?limit=500'));
-                var allActivitiesPromise = pluginRequest('GET', appendLaneParam('/api/plugin/activities?limit=500'));
-                var schedulerPromise = pluginRequest('GET', appendLaneParam('/api/plugin/scheduler-states'));
+                // The newest 500 rows cluster-wide may hold none of the selected job type on a busy
+                // cluster, so the job-type sub-tabs fetch their own server-side filtered window.
+                var scopedJobType = monitorJobTypeScope();
+                var sameScope = state.monitorRowsScope === scopedJobType;
+
+                // A failed request keeps its last rows; the rest still land and the failure is rethrown.
+                var failure = null;
+                function keepOnFailure(promise, previous) {
+                    return promise.catch(function(e) {
+                        failure = failure || e;
+                        return previous;
+                    });
+                }
+
+                var allJobsPromise = keepOnFailure(pluginRequest('GET', appendLaneParam('/api/plugin/jobs?limit=500')), state.allJobs);
+                var allActivitiesPromise = keepOnFailure(pluginRequest('GET', appendLaneParam('/api/plugin/activities?limit=500')), state.allActivities);
+                var jobsPromise = scopedJobType ? keepOnFailure(pluginRequest('GET', appendJobTypeParam(appendLaneParam('/api/plugin/jobs?limit=500'), scopedJobType)), sameScope ? state.jobs : null) : allJobsPromise;
+                var activitiesPromise = scopedJobType ? keepOnFailure(pluginRequest('GET', appendJobTypeParam(appendLaneParam('/api/plugin/activities?limit=500'), scopedJobType)), sameScope ? state.activities : null) : allActivitiesPromise;
+                var schedulerPromise = keepOnFailure(pluginRequest('GET', appendLaneParam('/api/plugin/scheduler-states')), state.schedulerStates);
                 var schedulerStatusPromise = pluginRequest('GET', appendLaneParam('/api/plugin/scheduler-status'));
 
                 var allJobs = await allJobsPromise;
                 var allActivities = await allActivitiesPromise;
+                var jobs = await jobsPromise;
+                var activities = await activitiesPromise;
                 var schedulerStates = await schedulerPromise;
                 var schedulerStatus = null;
                 try {
@@ -3089,11 +3130,15 @@ templ Plugin(page string, initialJob string, lane string) {
                 } catch (e) {
                     schedulerStatus = null;
                 }
+                if (monitorJobTypeScope() !== scopedJobType) {
+                    return;
+                }
 
-                state.jobs = Array.isArray(allJobs) ? allJobs : [];
-                state.activities = Array.isArray(allActivities) ? allActivities : [];
-                state.allJobs = state.jobs;
-                state.allActivities = state.activities;
+                state.allJobs = Array.isArray(allJobs) ? allJobs : [];
+                state.allActivities = Array.isArray(allActivities) ? allActivities : [];
+                state.jobs = Array.isArray(jobs) ? jobs : state.allJobs;
+                state.activities = Array.isArray(activities) ? activities : state.allActivities;
+                state.monitorRowsScope = scopedJobType;
                 state.schedulerStates = Array.isArray(schedulerStates) ? schedulerStates : [];
                 state.schedulerStatus = schedulerStatus && schedulerStatus.scheduler ? schedulerStatus.scheduler : null;
                 state.schedulerLock = schedulerStatus && schedulerStatus.lock ? schedulerStatus.lock : null;
@@ -3105,6 +3150,9 @@ templ Plugin(page string, initialJob string, lane string) {
                 renderSchedulerStatus();
                 renderStatus();
                 renderJobTypeSummary();
+                if (failure) {
+                    throw failure;
+                }
             }
 
             async function refreshAll() {
@@ -3133,8 +3181,13 @@ templ Plugin(page string, initialJob string, lane string) {
                     }
 
                     await refreshJobsAndActivities();
+                    state.refreshError = '';
                 } catch (e) {
-                    notify('Refresh failed: ' + e.message, 'error');
+                    // The 3s timer would otherwise reopen the same alert on every tick.
+                    if (e.message !== state.refreshError) {
+                        state.refreshError = e.message;
+                        notify('Refresh failed: ' + e.message, 'error');
+                    }
                 }
             }
 
@@ -3231,10 +3284,11 @@ templ Plugin(page string, initialJob string, lane string) {
                             }
                             await loadRunHistory(jobType);
                         }
-                        renderQueueJobs();
-                        renderDetectionJobs();
-                        renderExecutionJobs();
-                        renderExecutionActivities();
+                        try {
+                            await refreshJobsAndActivities();
+                        } catch (e) {
+                            notify('Refresh failed: ' + e.message, 'error');
+                        }
                         return;
                     }
 
@@ -3300,10 +3354,11 @@ templ Plugin(page string, initialJob string, lane string) {
                         console.error('Failed to load data on popstate navigation:', e);
                     }
                 }
-                renderQueueJobs();
-                renderDetectionJobs();
-                renderExecutionJobs();
-                renderExecutionActivities();
+                try {
+                    await refreshJobsAndActivities();
+                } catch (e) {
+                    console.error('Failed to refresh jobs on popstate navigation:', e);
+                }
             });
 
             window.addEventListener('beforeunload', function() {

weed/admin/view/app/plugin_templ.go

@@ -109,7 +109,7 @@ func Plugin(page string, initialJob string, lane string) templ.Component {
 		if templ_7745c5c3_Err != nil {
 			return templ_7745c5c3_Err
 		}
-		templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 7, "<div class=\"plugin-section plugin-section-overview\"><div class=\"row mb-4\"><div class=\"col-md-3 mb-3\"><div class=\"card border-info h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Workers</div><h4 class=\"mb-0\" id=\"plugin-status-workers\">0</h4></div></div></div><div class=\"col-md-3 mb-3\"><div class=\"card border-warning h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Active Jobs</div><h4 class=\"mb-0\" id=\"plugin-status-active-jobs\">0</h4></div></div></div><div class=\"col-md-3 mb-3\"><div class=\"card border-success h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Activities (recent)</div><h4 class=\"mb-0\" id=\"plugin-status-activities\">0</h4></div></div></div><div class=\"col-md-3 mb-3\"><div class=\"card border-secondary h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Next scheduled check</div><h4 class=\"mb-0 plugin-scheduler-next-run\">-</h4><div class=\"text-muted small plugin-scheduler-next-run-meta\"></div></div></div></div></div><div class=\"row mb-4\"><div class=\"col-12\"><div class=\"card shadow-sm\"><div class=\"card-header\"><h5 class=\"mb-0\"><i class=\"bi bi-bar-chart me-2\"></i>Per Job Type Summary</h5></div><div class=\"card-body p-0\"><div class=\"table-responsive\"><table class=\"table table-sm table-hover mb-0\"><thead class=\"table-light\"><tr><th>Job Type</th><th>Active Jobs</th><th>Recent Activities</th></tr></thead> <tbody id=\"plugin-jobtype-summary-body\"><tr><td colspan=\"3\" class=\"text-muted text-center py-3\">Loading...</td></tr></tbody></table></div></div></div></div></div><div class=\"row mb-4\"><div class=\"col-12\"><div class=\"card shadow-sm\"><div class=\"card-header d-flex justify-content-between align-items-center flex-wrap gap-2\"><h5 class=\"mb-0\"><i class=\"bi bi-clock me-2\"></i>Scheduler State</h5><small class=\"text-muted\">Sequential scheduler with per-job runtime limits</small></div><div class=\"card-body p-0\"><div class=\"table-responsive\"><table class=\"table table-sm table-hover mb-0\"><thead class=\"table-light\"><tr><th>Job Type</th><th>Enabled</th><th>Detector</th><th>In Flight</th><th>Max Runtime</th><th>Exec Global</th><th>Exec/Worker</th><th>Executor Workers</th><th>Effective Exec</th><th>Last Run</th></tr></thead> <tbody id=\"plugin-scheduler-table-body\"><tr><td colspan=\"10\" class=\"text-muted text-center py-3\">Loading...</td></tr></tbody></table></div></div></div></div></div><div class=\"row mb-4\"><div class=\"col-12\"><div class=\"card shadow-sm\"><div class=\"card-header\"><h5 class=\"mb-0\"><i class=\"bi bi-people me-2\"></i>Workers</h5></div><div class=\"card-body p-0\"><div class=\"table-responsive\" style=\"max-height: 560px; overflow-y: auto;\"><table class=\"table table-sm table-hover mb-0\"><thead class=\"table-light sticky-top\"><tr><th>Worker</th><th>Address</th><th>Capabilities</th><th>Load</th></tr></thead> <tbody id=\"plugin-workers-table-body\"><tr><td colspan=\"4\" class=\"text-muted text-center py-3\">Loading...</td></tr></tbody></table></div></div></div></div></div></div><div class=\"plugin-section plugin-section-configuration\"><div class=\"row mb-4\"><div class=\"col-lg-8 mb-3\"><div class=\"card shadow-sm h-100\"><div class=\"card-header d-flex justify-content-between align-items-center\"><h5 class=\"mb-0\"><i class=\"bi bi-sliders me-2\"></i>Job Type Configuration</h5><small class=\"text-muted\" id=\"plugin-config-updated-at\">Not loaded</small></div><div class=\"card-body\"><div class=\"mb-3\"><div class=\"small text-muted\">Selected Job Type</div><div class=\"fw-semibold\" id=\"plugin-selected-job-type\">-</div></div><div class=\"mb-3\"><div class=\"d-flex justify-content-between align-items-center mb-2\"><h6 class=\"mb-0\">Descriptor</h6><button type=\"button\" class=\"btn btn-sm btn-outline-secondary\" id=\"plugin-refresh-schema-btn\"><i class=\"bi bi-cloud-download me-1\"></i>Refresh Schema</button></div><div class=\"alert alert-light border\" id=\"plugin-descriptor-summary\">Select a job type to load schema and config.</div></div><div class=\"mb-3\"><h6>Admin Config Form</h6><div id=\"plugin-admin-config-form\" class=\"plugin-form-root\"><div class=\"text-muted\">No admin form loaded.</div></div></div><div class=\"mb-3\"><h6>Worker Config Form</h6><div id=\"plugin-worker-config-form\" class=\"plugin-form-root\"><div class=\"text-muted\">No worker form loaded.</div></div></div><div class=\"d-flex flex-wrap gap-2\"><button type=\"button\" class=\"btn btn-primary\" id=\"plugin-save-config-btn\"><i class=\"bi bi-floppy me-1\"></i>Save Config</button> <button type=\"button\" class=\"btn btn-outline-primary\" id=\"plugin-trigger-detection-btn\"><i class=\"bi bi-search me-1\"></i>Run Detection</button> <button type=\"button\" class=\"btn btnLine truncated
+		templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 7, "<div class=\"plugin-section plugin-section-overview\"><div class=\"row mb-4\"><div class=\"col-md-3 mb-3\"><div class=\"card border-info h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Workers</div><h4 class=\"mb-0\" id=\"plugin-status-workers\">0</h4></div></div></div><div class=\"col-md-3 mb-3\"><div class=\"card border-warning h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Active Jobs</div><h4 class=\"mb-0\" id=\"plugin-status-active-jobs\">0</h4></div></div></div><div class=\"col-md-3 mb-3\"><div class=\"card border-success h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Activities (recent)</div><h4 class=\"mb-0\" id=\"plugin-status-activities\">0</h4></div></div></div><div class=\"col-md-3 mb-3\"><div class=\"card border-secondary h-100\"><div class=\"card-body text-center\"><div class=\"text-uppercase text-muted small mb-1\">Next scheduled check</div><h4 class=\"mb-0 plugin-scheduler-next-run\">-</h4><div class=\"text-muted small plugin-scheduler-next-run-meta\"></div></div></div></div></div><div class=\"row mb-4\"><div class=\"col-12\"><div class=\"card shadow-sm\"><div class=\"card-header\"><h5 class=\"mb-0\"><i class=\"bi bi-bar-chart me-2\"></i>Per Job Type Summary</h5></div><div class=\"card-body p-0\"><div class=\"table-responsive\"><table class=\"table table-sm table-hover mb-0\"><thead class=\"table-light\"><tr><th>Job Type</th><th>Active Jobs</th><th>Recent Activities</th></tr></thead> <tbody id=\"plugin-jobtype-summary-body\"><tr><td colspan=\"3\" class=\"text-muted text-center py-3\">Loading...</td></tr></tbody></table></div></div></div></div></div><div class=\"row mb-4\"><div class=\"col-12\"><div class=\"card shadow-sm\"><div class=\"card-header d-flex justify-content-between align-items-center flex-wrap gap-2\"><h5 class=\"mb-0\"><i class=\"bi bi-clock me-2\"></i>Scheduler State</h5><small class=\"text-muted\">Sequential scheduler with per-job runtime limits</small></div><div class=\"card-body p-0\"><div class=\"table-responsive\"><table class=\"table table-sm table-hover mb-0\"><thead class=\"table-light\"><tr><th>Job Type</th><th>Enabled</th><th>Detector</th><th>In Flight</th><th>Max Runtime</th><th>Exec Global</th><th>Exec/Worker</th><th>Executor Workers</th><th>Effective Exec</th><th>Last Run</th></tr></thead> <tbody id=\"plugin-scheduler-table-body\"><tr><td colspan=\"10\" class=\"text-muted text-center py-3\">Loading...</td></tr></tbody></table></div></div></div></div></div><div class=\"row mb-4\"><div class=\"col-12\"><div class=\"card shadow-sm\"><div class=\"card-header\"><h5 class=\"mb-0\"><i class=\"bi bi-people me-2\"></i>Workers</h5></div><div class=\"card-body p-0\"><div class=\"table-responsive\" style=\"max-height: 560px; overflow-y: auto;\"><table class=\"table table-sm table-hover mb-0\"><thead class=\"table-light sticky-top\"><tr><th>Worker</th><th>Address</th><th>Capabilities</th><th>Load</th></tr></thead> <tbody id=\"plugin-workers-table-body\"><tr><td colspan=\"4\" class=\"text-muted text-center py-3\">Loading...</td></tr></tbody></table></div></div></div></div></div></div><div class=\"plugin-section plugin-section-configuration\"><div class=\"row mb-4\"><div class=\"col-lg-8 mb-3\"><div class=\"card shadow-sm h-100\"><div class=\"card-header d-flex justify-content-between align-items-center\"><h5 class=\"mb-0\"><i class=\"bi bi-sliders me-2\"></i>Job Type Configuration</h5><small class=\"text-muted\" id=\"plugin-config-updated-at\">Not loaded</small></div><div class=\"card-body\"><div class=\"mb-3\"><div class=\"small text-muted\">Selected Job Type</div><div class=\"fw-semibold\" id=\"plugin-selected-job-type\">-</div></div><div class=\"mb-3\"><div class=\"d-flex justify-content-between align-items-center mb-2\"><h6 class=\"mb-0\">Descriptor</h6><button type=\"button\" class=\"btn btn-sm btn-outline-secondary\" id=\"plugin-refresh-schema-btn\"><i class=\"bi bi-cloud-download me-1\"></i>Refresh Schema</button></div><div class=\"alert alert-light border\" id=\"plugin-descriptor-summary\">Select a job type to load schema and config.</div></div><div class=\"mb-3\"><h6>Admin Config Form</h6><div id=\"plugin-admin-config-form\" class=\"plugin-form-root\"><div class=\"text-muted\">No admin form loaded.</div></div></div><div class=\"mb-3\"><h6>Worker Config Form</h6><div id=\"plugin-worker-config-form\" class=\"plugin-form-root\"><div class=\"text-muted\">No worker form loaded.</div></div></div><div class=\"d-flex flex-wrap gap-2\"><button type=\"button\" class=\"btn btn-primary\" id=\"plugin-save-config-btn\"><i class=\"bi bi-floppy me-1\"></i>Save Config</button> <button type=\"button\" class=\"btn btn-outline-primary\" id=\"plugin-trigger-detection-btn\"><i class=\"bi bi-search me-1\"></i>Run Detection</button> <button type=\"button\" class=\"btn btnLine truncated
 		if templ_7745c5c3_Err != nil {
 			return templ_7745c5c3_Err
 		}

weed/command/filer.go

@@ -506,12 +506,9 @@ func (fo *FilerOptions) startFiler() {
 	grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.filer"))
 	filer_pb.RegisterSeaweedFilerServer(grpcS, fs)
 
-	// Register the IAM gRPC service. Auth is opt-in: when
-	// jwt.filer_signing.key is configured the service requires a Bearer token
-	// signed with that key; otherwise it runs unauthenticated, matching the
-	// rest of the filer's gRPC surface. Operators who expose the filer gRPC
-	// port beyond a trusted network should set jwt.filer_signing.key on both
-	// the filer and the admin server.
+	// IAM RPCs require a Bearer token signed with jwt.filer_signing.key when it
+	// is set. The key does not protect the rest of this port; that needs
+	// [grpc.filer] mTLS.
 	if credentialManager != nil {
 		adminSigningKey := security.SigningKey(util.GetViper().GetString("jwt.filer_signing.key"))
 		iamGrpcServer := weed_server.NewIamGrpcServer(credentialManager, adminSigningKey)

weed/command/scaffold/security.toml

@@ -57,9 +57,11 @@ expires_after_seconds = 10           # seconds
 #       CreateAccessKey, ...). When set, every IAM RPC must carry a Bearer
 #       token signed with this key in its "authorization" gRPC metadata; mint
 #       such a token with security.GenJwtForFilerAdmin. When empty, the IAM
-#       gRPC service runs unauthenticated, like the rest of the filer's gRPC
-#       surface — set the key on both filer and admin if the gRPC port is
-#       reachable beyond a trusted network.
+#       gRPC service runs unauthenticated and refuses its OIDC provider and
+#       role RPCs. When set, set it on both filer and admin.
+#       The key does not protect the other filer gRPC RPCs, which can read and
+#       write the IAM store under /etc/iam; if the filer gRPC port is reachable
+#       beyond a trusted network, enable [grpc.filer] mTLS below.
 # the jwt defaults to expire after 10 seconds.
 [jwt.filer_signing]
 key = ""

weed/s3api/direct_read_fixture_test.go

@@ -0,0 +1,159 @@
+package s3api
+
+import (
+	"bytes"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"io"
+	"os"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// Writes a known-answer fixture for readers that decode SSE-S3 objects and
+// filer-encrypted chunks outside the gateway, from this package's own
+// encryption: DIRECT_READ_FIXTURE=/path go test -run TestGenerateDirectReadFixture
+func TestGenerateDirectReadFixture(t *testing.T) {
+	out := os.Getenv("DIRECT_READ_FIXTURE")
+	if out == "" {
+		t.Skip("set DIRECT_READ_FIXTURE to write the fixture")
+	}
+	kek, err := deriveKeyFromSecret("s3vectors-test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	globalSSES3KeyManager.mu.Lock()
+	previous := globalSSES3KeyManager.superKey
+	globalSSES3KeyManager.superKey = kek
+	globalSSES3KeyManager.mu.Unlock()
+	t.Cleanup(func() {
+		globalSSES3KeyManager.mu.Lock()
+		globalSSES3KeyManager.superKey = previous
+		globalSSES3KeyManager.mu.Unlock()
+	})
+
+	plain := make([]byte, 5000)
+	for i := range plain {
+		plain[i] = byte(i*7 + i/13)
+	}
+	key, err := GenerateSSES3Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	baseIV := []byte("0123456789abcdef")
+
+	type chunk struct {
+		Offset      int64  `json:"offset"`
+		Ciphertext  string `json:"ciphertext"`
+		SseMetadata string `json:"sse_metadata"`
+	}
+	var chunks []chunk
+	for _, part := range [][2]int64{{0, 2048}, {2048, 2952}} {
+		off, size := part[0], part[1]
+		enc, iv, err := CreateSSES3EncryptedReaderWithBaseIV(bytes.NewReader(plain[off:off+size]), key, baseIV, off)
+		if err != nil {
+			t.Fatal(err)
+		}
+		ct, err := io.ReadAll(enc)
+		if err != nil {
+			t.Fatal(err)
+		}
+		meta, err := SerializeSSES3Metadata(&SSES3Key{Key: key.Key, KeyID: key.KeyID, Algorithm: key.Algorithm, IV: iv})
+		if err != nil {
+			t.Fatal(err)
+		}
+		chunks = append(chunks, chunk{off, hex.EncodeToString(ct), string(meta)})
+	}
+	objectMeta, err := SerializeSSES3Metadata(&SSES3Key{Key: key.Key, KeyID: key.KeyID, Algorithm: key.Algorithm, IV: baseIV})
+	if err != nil {
+		t.Fatal(err)
+	}
+	encWhole, _, err := CreateSSES3EncryptedReaderWithBaseIV(bytes.NewReader(plain), key, baseIV, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	whole, err := io.ReadAll(encWhole)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ck := util.GenCipherKey()
+	cipherPlain, err := util.Encrypt(plain, ck)
+	if err != nil {
+		t.Fatal(err)
+	}
+	gz, err := util.GzipData(plain)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cipherGzip, err := util.Encrypt(gz, ck)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	km := NewSSES3KeyManager("pass-phrase")
+	v2, err := km.wrapKEK(kek)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wrappingKey, err := km.deriveWrappingKey(legacyKEKWrappingSalt)
+	if err != nil {
+		t.Fatal(err)
+	}
+	block, err := aes.NewCipher(wrappingKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	gcm, err := cipher.NewGCM(block)
+	if err != nil {
+		t.Fatal(err)
+	}
+	nonce := make([]byte, gcm.NonceSize())
+	if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
+		t.Fatal(err)
+	}
+	v1 := base64.StdEncoding.EncodeToString(gcm.Seal(nonce, nonce, kek, nil))
+	for _, wrapped := range []struct {
+		data []byte
+		isV2 bool
+	}{{[]byte(v1), false}, {v2, true}} {
+		unwrapped, isV2, err := km.unwrapKEK(wrapped.data)
+		if err != nil || isV2 != wrapped.isV2 || !bytes.Equal(unwrapped, kek) {
+			t.Fatalf("unwrapKEK(v2=%v) = v2=%v, err=%v; want the KEK back", wrapped.isV2, isV2, err)
+		}
+	}
+
+	fixture := map[string]any{
+		"kek_secret":        "s3vectors-test",
+		"kek_hex":           hex.EncodeToString(kek),
+		"kek_passphrase":    "pass-phrase",
+		"kek_wrapped_v1":    v1,
+		"kek_wrapped_v2":    string(v2),
+		"plaintext":         hex.EncodeToString(plain),
+		"object_metadata":   string(objectMeta),
+		"chunks":            chunks,
+		"whole_ciphertext":  hex.EncodeToString(whole),
+		"cipher_key":        hex.EncodeToString(ck),
+		"cipher_plain":      hex.EncodeToString(cipherPlain),
+		"cipher_gzip":       hex.EncodeToString(cipherGzip),
+		"commitment_sample": hex.EncodeToString(ComputeKeyCommitment(key.Key, baseIV, key.Algorithm)),
+	}
+	f, err := os.Create(out)
+	if err != nil {
+		t.Fatal(err)
+	}
+	enc := json.NewEncoder(f)
+	enc.SetIndent("", " ")
+	if err := enc.Encode(fixture); err != nil {
+		f.Close()
+		t.Fatal(err)
+	}
+	if err := f.Close(); err != nil {
+		t.Fatal(err)
+	}
+}

weed/s3api/s3_metadata_util.go

@@ -25,6 +25,10 @@ func ParseS3Metadata(r *http.Request, existing map[string][]byte, isReplace bool
 
 	// Storage class
 	if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" {
+		if !validateStorageClass(sc) {
+			glog.Warningf("Invalid storage class %q", sc)
+			return nil, s3err.ErrInvalidStorageClass
+		}
 		metadata[s3_constants.AmzStorageClass] = []byte(sc)
 	}
 

weed/s3api/s3_sse_s3_kek_passphrase_test.go

@@ -87,3 +87,25 @@ func TestNoPassphraseKeepsLegacyHexDecodePath(t *testing.T) {
 		t.Fatal("wrapKEK without passphrase should fail; got nil error")
 	}
 }
+
+// A KEK wrapped in the legacy v1 layout (nonce || ciphertext+tag) under the
+// fixed salt "seaweedfs-sse-s3-kek-wrapping-v1", produced outside this
+// package, so a change to legacyKEKWrappingSalt cannot pass by changing
+// both sides.
+func TestUnwrapKEKReadsPinnedLegacyV1(t *testing.T) {
+	km := NewSSES3KeyManager("legacy-passphrase")
+	got, isV2, err := km.unwrapKEK([]byte("oKGio6SlpqeoqaqrixOn/dNaewpX3OoMjLT3VUFh+zg+Hd1fqbiWa6Xhc3SAg6hZdXb+Pwo520Fc1dfj"))
+	if err != nil {
+		t.Fatalf("unwrapKEK: %v", err)
+	}
+	if isV2 {
+		t.Fatal("a legacy payload was reported as v2")
+	}
+	want := make([]byte, SSES3KeySize)
+	for i := range want {
+		want[i] = byte(i)
+	}
+	if string(got) != string(want) {
+		t.Fatalf("unwrapped %x, want %x", got, want)
+	}
+}

weed/s3api/s3api_object_handlers_multipart.go

@@ -389,6 +389,11 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	if _, errCode := validatePutHeaders(r); errCode != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, errCode)
+		return
+	}
+
 	dataReader, s3ErrCode := getRequestDataReader(s3a, r)
 	if s3ErrCode != s3err.ErrNone {
 		glog.Errorf("PutObjectPartHandler: getRequestDataReader failed with code %v", s3ErrCode)

weed/s3api/s3api_object_handlers_put.go

@@ -134,6 +134,12 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	tagSet, headerCode := validatePutHeaders(r)
+	if headerCode != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, headerCode)
+		return
+	}
+
 	if r.Header.Get("Cache-Control") != "" {
 		if _, err = cacheobject.ParseRequestCacheControl(r.Header.Get("Cache-Control")); err != nil {
 			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest)
@@ -279,6 +285,12 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 						entry.Extended = make(map[string][]byte)
 					}
 					entry.Extended[s3_constants.ExtETagKey] = []byte(dirEtag)
+					if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" {
+						entry.Extended[s3_constants.AmzStorageClass] = []byte(sc)
+					}
+					for key, value := range tagSet {
+						entry.Extended[S3TAG_PREFIX+key] = []byte(value)
+					}
 
 					storeSSEMetadata(entry, sseResult)
 
@@ -524,6 +536,28 @@ func withMultipartPart(r *http.Request) *http.Request {
 	return r.WithContext(context.WithValue(r.Context(), multipartPartContextKey{}, true))
 }
 
+// validatePutHeaders checks x-amz-storage-class and x-amz-tagging, so the
+// handlers can reject them before reading the body and putToFiler before
+// uploading any chunk. It returns the parsed tag set.
+func validatePutHeaders(r *http.Request) (map[string]string, s3err.ErrorCode) {
+	if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" && !validateStorageClass(sc) {
+		return nil, s3err.ErrInvalidStorageClass
+	}
+	tagging := r.Header.Get(s3_constants.AmzObjectTagging)
+	if tagging == "" {
+		return nil, s3err.ErrNone
+	}
+	tagSet, err := parseTagsHeader(tagging)
+	if err == nil {
+		err = ValidateTags(tagSet)
+	}
+	if err != nil {
+		glog.V(1).Infof("invalid tagging header %q: %v", tagging, err)
+		return nil, s3err.ErrInvalidTag
+	}
+	return tagSet, s3err.ErrNone
+}
+
 func isMultipartPartRequest(r *http.Request) bool {
 	marked, _ := r.Context().Value(multipartPartContextKey{}).(bool)
 	return marked
@@ -546,21 +580,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 		return "", s3err.ErrInvalidRequest, SSEResponseMetadata{}
 	}
 
-	// Reject invalid tags before any chunks are uploaded so a failed request
-	// leaves nothing behind on the volume servers.
-	var tagSet map[string]string
-	if tagging := r.Header.Get(s3_constants.AmzObjectTagging); tagging != "" {
-		parsed, err := parseTagsHeader(tagging)
-		if err != nil {
-			glog.Warningf("putToFiler: invalid tagging header for %s: %v", filePath, err)
-			return "", s3err.ErrInvalidTag, SSEResponseMetadata{}
-		}
-		if err := ValidateTags(parsed); err != nil {
-			glog.Warningf("putToFiler: invalid tags (%d) for %s: %v", len(parsed), filePath, err)
-			return "", s3err.ErrInvalidTag, SSEResponseMetadata{}
-		}
-		tagSet = parsed
+	tagSet, headerCode := validatePutHeaders(r)
+	if headerCode != s3err.ErrNone {
+		return "", headerCode, SSEResponseMetadata{}
 	}
+	storageClass := r.Header.Get(s3_constants.AmzStorageClass)
 	// NEW OPTIMIZATION: Write directly to volume servers, bypassing filer proxy
 	// This eliminates the filer proxy overhead for PUT operations
 	// Note: filePath is now passed directly instead of URL (no parsing needed)
@@ -963,13 +987,9 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 		}
 	}
 
-	// Store the storage class from header
-	if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" {
-		if !validateStorageClass(sc) {
-			glog.Warningf("putToFiler: Invalid storage class '%s' for %s", sc, filePath)
-			return "", s3err.ErrInvalidStorageClass, SSEResponseMetadata{}
-		}
-		entry.Extended[s3_constants.AmzStorageClass] = []byte(sc)
+	// Store the storage class validated before the chunk upload
+	if storageClass != "" {
+		entry.Extended[s3_constants.AmzStorageClass] = []byte(storageClass)
 	}
 
 	// Store the tags validated before the chunk upload

weed/s3api/s3api_storage_class_validation_test.go

@@ -0,0 +1,262 @@
+package s3api
+
+import (
+	"encoding/xml"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+	"github.com/stretchr/testify/require"
+)
+
+func (f *fakeVolumeServer) storedCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.stored)
+}
+
+// An invalid x-amz-storage-class must be rejected before the body is
+// uploaded, so a failed PUT or UploadPart leaves no chunks behind.
+func TestPutToFilerStorageClass(t *testing.T) {
+	for _, tt := range []struct {
+		name       string
+		class      string
+		part       bool
+		wantCode   s3err.ErrorCode
+		wantStored int
+	}{
+		{name: "invalid", class: "BOGUS", wantCode: s3err.ErrInvalidStorageClass},
+		{name: "invalid on a part", class: "BOGUS", part: true, wantCode: s3err.ErrInvalidStorageClass},
+		{name: "valid", class: "STANDARD_IA", wantCode: s3err.ErrNone, wantStored: 1},
+		{name: "absent", wantCode: s3err.ErrNone, wantStored: 1},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			volume := startFakeVolumeServer(t)
+			filer := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}}
+			s3a := newPutTestServer(t, startFakeFiler(t, filer))
+
+			r := httptest.NewRequest(http.MethodPut, "/b/o", nil)
+			if tt.class != "" {
+				r.Header.Set(s3_constants.AmzStorageClass, tt.class)
+			}
+			filePath, object, dest := "/buckets/b/o", "o", ""
+			if tt.part {
+				r = withMultipartPart(r)
+				filePath, object, dest = "/buckets/b/.uploads/u/0001.part", "", "/buckets/b/o"
+			}
+			_, code, _ := s3a.putToFiler(r, filePath, strings.NewReader("hello world"), "b", object, 1, 0, nil, false, dest)
+
+			require.Equal(t, tt.wantCode, code)
+			require.Equal(t, tt.wantStored, volume.storedCount(), "chunks stored on the volume server")
+			if tt.wantCode != s3err.ErrNone {
+				require.Empty(t, filer.entries)
+				return
+			}
+			entry := filer.entries["/buckets/b/o"]
+			require.NotNil(t, entry)
+			require.Equal(t, tt.class, string(entry.Extended[s3_constants.AmzStorageClass]))
+		})
+	}
+}
+
+// countingBody counts the reads a handler makes of the request body.
+type countingBody struct {
+	r     io.Reader
+	reads atomic.Int32
+}
+
+func (b *countingBody) Read(p []byte) (int, error) { b.reads.Add(1); return b.r.Read(p) }
+func (b *countingBody) Close() error               { return nil }
+
+// drainProbe is the ResponseWriter of these tests. WriteErrorResponse drains
+// the body before it answers, which would count as reads, so the probe notes
+// how many the handler had made when that drain starts.
+type drainProbe struct {
+	*httptest.ResponseRecorder
+	body         *countingBody
+	readsAtDrain int32
+}
+
+func (p *drainProbe) SetReadDeadline(time.Time) error {
+	if p.readsAtDrain < 0 {
+		p.readsAtDrain = p.body.reads.Load()
+	}
+	return nil
+}
+
+func newHeaderTestServer(t *testing.T, versioning string) (*S3ApiServer, *ambiguousPutFiler, *fakeVolumeServer) {
+	t.Helper()
+	util_http.InitGlobalHttpClient()
+	volume := startFakeVolumeServer(t)
+	filer := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filer))
+	s3a.iam = &IdentityAccessManagement{}
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: versioning})
+	return s3a, filer, volume
+}
+
+// newHeaderTestRequest builds a request without a Content-Type, so a handler
+// has to sniff one from the body.
+func newHeaderTestRequest(target, object, payload, header, value string) (*http.Request, *drainProbe) {
+	body := &countingBody{r: strings.NewReader(payload)}
+	req := httptest.NewRequest(http.MethodPut, target, body)
+	req.ContentLength = int64(len(payload))
+	req.Header.Set(header, value)
+	req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": object})
+	return req, &drainProbe{ResponseRecorder: httptest.NewRecorder(), body: body, readsAtDrain: -1}
+}
+
+func tagHeader(n int) string {
+	tags := make([]string, n)
+	for i := range tags {
+		tags[i] = fmt.Sprintf("k%d=v", i)
+	}
+	return strings.Join(tags, "&")
+}
+
+var invalidPutHeaders = []struct {
+	name, header, value string
+	want                s3err.ErrorCode
+}{
+	{"storage class", s3_constants.AmzStorageClass, "BOGUS", s3err.ErrInvalidStorageClass},
+	{"11 tags", s3_constants.AmzObjectTagging, tagHeader(11), s3err.ErrInvalidTag},
+}
+
+func requireRejected(t *testing.T, probe *drainProbe, volume *fakeVolumeServer, want s3err.ErrorCode) {
+	t.Helper()
+	apiErr := s3err.GetAPIError(want)
+	require.Equal(t, apiErr.HTTPStatusCode, probe.Code, probe.Body.String())
+	require.Contains(t, probe.Body.String(), "<Code>"+apiErr.Code+"</Code>")
+	require.Zero(t, probe.readsAtDrain, "body reads before the error response")
+	require.Zero(t, volume.storedCount(), "chunks stored on the volume server")
+}
+
+// An invalid storage class or tag set is answered before the handler touches
+// the body, whichever way the bucket versions: the content-type sniff of a PUT
+// without Content-Type blocks on the first read.
+func TestPutObjectRejectsInvalidHeadersBeforeReadingBody(t *testing.T) {
+	for _, state := range []struct{ name, versioning string }{
+		{"unversioned", ""},
+		{"versioning enabled", s3_constants.VersioningEnabled},
+		{"versioning suspended", s3_constants.VersioningSuspended},
+	} {
+		for _, h := range invalidPutHeaders {
+			t.Run(state.name+"/"+h.name, func(t *testing.T) {
+				s3a, filer, volume := newHeaderTestServer(t, state.versioning)
+				req, probe := newHeaderTestRequest("/b/o", "o", "hello world", h.header, h.value)
+
+				s3a.PutObjectHandler(probe, req)
+
+				requireRejected(t, probe, volume, h.want)
+				filer.mu.Lock()
+				defer filer.mu.Unlock()
+				require.Empty(t, filer.entries)
+			})
+		}
+	}
+}
+
+func TestPutObjectPartRejectsInvalidHeadersBeforeReadingBody(t *testing.T) {
+	for _, h := range invalidPutHeaders {
+		t.Run(h.name, func(t *testing.T) {
+			s3a, filer, volume := newHeaderTestServer(t, "")
+			uploadID := s3a.generateUploadID("o")
+			filer.mu.Lock()
+			filer.entries["/buckets/b/.uploads/"+uploadID] = &filer_pb.Entry{
+				Name: uploadID, IsDirectory: true,
+				Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("o")},
+			}
+			filer.mu.Unlock()
+			req, probe := newHeaderTestRequest("/b/o?partNumber=1&uploadId="+uploadID, "o", "hello world", h.header, h.value)
+
+			s3a.PutObjectPartHandler(probe, req)
+
+			requireRejected(t, probe, volume, h.want)
+		})
+	}
+}
+
+// A directory marker is validated like any other PUT.
+func TestPutObjectDirectoryMarkerRejectsInvalidHeaders(t *testing.T) {
+	for _, h := range invalidPutHeaders {
+		t.Run(h.name, func(t *testing.T) {
+			s3a, filer, volume := newHeaderTestServer(t, "")
+			req, probe := newHeaderTestRequest("/b/dir/", "dir/", "", h.header, h.value)
+
+			s3a.PutObjectHandler(probe, req)
+
+			requireRejected(t, probe, volume, h.want)
+			filer.mu.Lock()
+			defer filer.mu.Unlock()
+			require.Nil(t, filer.entries["/buckets/b/dir"], "directory marker created")
+		})
+	}
+}
+
+func TestPutObjectDirectoryMarkerStoresHeaders(t *testing.T) {
+	s3a, filer, _ := newHeaderTestServer(t, "")
+	req, probe := newHeaderTestRequest("/b/dir/", "dir/", "", s3_constants.AmzStorageClass, "STANDARD_IA")
+	req.Header.Set(s3_constants.AmzObjectTagging, "k1=v1&k2=v2")
+
+	s3a.PutObjectHandler(probe, req)
+
+	require.Equal(t, http.StatusOK, probe.Code, probe.Body.String())
+	filer.mu.Lock()
+	entry := filer.entries["/buckets/b/dir"]
+	require.NotNil(t, entry)
+	require.Equal(t, "STANDARD_IA", string(entry.Extended[s3_constants.AmzStorageClass]))
+	require.Equal(t, "v1", string(entry.Extended[S3TAG_PREFIX+"k1"]))
+	require.Equal(t, "v2", string(entry.Extended[S3TAG_PREFIX+"k2"]))
+	filer.mu.Unlock()
+
+	getReq := httptest.NewRequest(http.MethodGet, "/b/dir/?tagging", nil)
+	getReq = mux.SetURLVars(getReq, map[string]string{"bucket": "b", "object": "dir/"})
+	rec := httptest.NewRecorder()
+	s3a.GetObjectTaggingHandler(rec, getReq)
+	require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
+	var tagging Tagging
+	require.NoError(t, xml.Unmarshal(rec.Body.Bytes(), &tagging))
+	require.ElementsMatch(t, []Tag{{Key: "k1", Value: "v1"}, {Key: "k2", Value: "v2"}}, tagging.TagSet.Tag)
+}
+
+// A valid storage class still stores, and the content type is still sniffed.
+func TestPutObjectValidStorageClassSniffsContentType(t *testing.T) {
+	s3a, filer, volume := newHeaderTestServer(t, "")
+	req, probe := newHeaderTestRequest("/b/o", "o", "hello world", s3_constants.AmzStorageClass, "STANDARD_IA")
+
+	s3a.PutObjectHandler(probe, req)
+
+	require.Equal(t, http.StatusOK, probe.Code, probe.Body.String())
+	require.Equal(t, 1, volume.storedCount())
+	filer.mu.Lock()
+	defer filer.mu.Unlock()
+	entry := filer.entries["/buckets/b/o"]
+	require.NotNil(t, entry)
+	require.Equal(t, "STANDARD_IA", string(entry.Extended[s3_constants.AmzStorageClass]))
+	require.Equal(t, "text/plain; charset=utf-8", entry.Attributes.Mime)
+}
+
+// CreateMultipartUpload stores the storage class for the completed object,
+// so it must reject an invalid one as PutObject and CopyObject do.
+func TestParseS3MetadataRejectsInvalidStorageClass(t *testing.T) {
+	req := httptest.NewRequest(http.MethodPost, "http://s3/b/o?uploads", nil)
+	req.Header.Set(s3_constants.AmzStorageClass, "BOGUS")
+	_, code := ParseS3Metadata(req, nil, false)
+	require.Equal(t, s3err.ErrInvalidStorageClass, code)
+
+	req.Header.Set(s3_constants.AmzStorageClass, "GLACIER")
+	metadata, code := ParseS3Metadata(req, nil, false)
+	require.Equal(t, s3err.ErrNone, code)
+	require.Equal(t, "GLACIER", string(metadata[s3_constants.AmzStorageClass]))
+}

weed/server/filer_server_handlers_iam_grpc.go

@@ -17,11 +17,11 @@ import (
 )
 
 // IamGrpcServer implements the IAM gRPC service on the filer.
-// Auth is opt-in: when jwt.filer_signing.key is set in security.toml the
-// service requires a Bearer token in the "authorization" metadata signed with
-// that key; when it is empty every RPC is accepted unauthenticated, matching
-// the rest of SeaweedFS's gRPC surface. Operators who expose the filer gRPC
-// port beyond a trusted network should configure the key.
+// When jwt.filer_signing.key is set, every RPC requires a Bearer token signed
+// with it; otherwise RPCs run unauthenticated, except the OIDC provider and
+// role RPCs, which are refused. The key gates only this service: the filer RPCs
+// on the same port can still read and write the IAM store under /etc/iam.
+// Protect a filer gRPC port reachable by untrusted clients with [grpc.filer] mTLS.
 type IamGrpcServer struct {
 	iam_pb.UnimplementedSeaweedIdentityAccessManagementServer
 	credentialManager *credential.CredentialManager

weed/shell/command_fs_distribute_chunks.go

@@ -23,6 +23,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+	"google.golang.org/protobuf/proto"
 )
 
 func init() {
@@ -116,6 +117,7 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
 	if err != nil {
 		return fmt.Errorf("lookup %s: %v", path, err)
 	}
+	readEntry := proto.Clone(entry).(*filer_pb.Entry)
 
 	chunks, hadManifest, err := resolveEntryDataChunks(commandEnv, entry)
 	if err != nil {
@@ -180,6 +182,11 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
 	fmt.Fprintf(writer, "\nExecuting redistribution...\n")
 	defer util_http.GetGlobalHttpClient().CloseIdleConnections()
 
+	readFids := make(map[string]bool, len(chunks))
+	for _, chunk := range chunks {
+		readFids[chunk.GetFileIdString()] = true
+	}
+
 	movedCount, err := executeChunkMoves(commandEnv, writer, moves, volumeInfoMap)
 	if err != nil {
 		return err
@@ -191,8 +198,14 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
 
 	finalChunks := chunks
 	if hadManifest {
-		// no chunk deleter here: a failed fold reports the blobs it saved
-		remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), nil, chunks)
+		dropSaved := func(saved []*filer_pb.FileChunk) {
+			var blobs []orphanedNeedle
+			for _, chunk := range saved {
+				blobs = append(blobs, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: chunk.GetFileIdString()})
+			}
+			deleteOrphanedNeedles(commandEnv, util.FullPath(path), blobs, true)
+		}
+		remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), dropSaved, chunks)
 		if manErr != nil {
 			fmt.Fprintf(writer, "WARNING: re-manifestize failed: %v. Writing flat chunk list.\n", manErr)
 		} else {
@@ -203,12 +216,24 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr
 	entry.Chunks = finalChunks
 
 	fmt.Fprintf(writer, "\nUpdating filer metadata (%d chunks moved)...\n", movedCount)
+	var stale bool
 	err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
-			Directory: dir,
-			Entry:     entry,
-		})
+		var uErr error
+		stale, uErr = updateEntryIfUnchanged(context.Background(), client, dir, entry, readEntry)
+		return uErr
 	})
+	if stale {
+		var copies []orphanedNeedle
+		for _, chunk := range slices.Concat(chunks, finalChunks) {
+			if fid := chunk.GetFileIdString(); !readFids[fid] {
+				readFids[fid] = true
+				copies = append(copies, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: fid})
+			}
+		}
+		deleteOrphanedNeedles(commandEnv, util.FullPath(path), copies, true)
+		fmt.Fprintf(writer, "%s was changed or deleted during redistribution; the filer kept that version and the moved copies were deleted.\n", path)
+		return err
+	}
 	if err != nil {
 		fmt.Fprintf(writer, "FAILED to update filer metadata: %v\n", err)
 		fmt.Fprintf(writer, "WARNING: New chunks were uploaded but metadata was not updated.\n")

weed/shell/command_fs_merge_volumes.go

@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"sort"
 	"strconv"
@@ -21,6 +22,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"golang.org/x/exp/maps"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/seaweedfs/seaweedfs/weed/operation"
@@ -54,6 +57,9 @@ func (c *commandFsMergeVolumes) Help() string {
 	-toVolumeId accepts a comma-separated list. With -fromVolumeId, the source
 	chunks are distributed across the listed volumes by remaining capacity, so a
 	volume that does not fit into any single target can still be cleared.
+
+	-apply needs the shell lock (run "lock" first), so two runs never move the
+	same chunks at once.
 `
 }
 
@@ -72,6 +78,11 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 	if err = fsMergeVolumesCommand.Parse(args); err != nil {
 		return err
 	}
+	if *apply {
+		if err = commandEnv.confirmIsLocked(args); err != nil {
+			return err
+		}
+	}
 
 	dir := *dirArg
 	if dir != "/" {
@@ -171,7 +182,16 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 			// look like mergeVolumes hadn't done anything. Track the old fids
 			// and delete them below after the filer update commits, so the
 			// filer never points at a fid we already deleted.
-			var movedSources []orphanedNeedle
+			var moved rewrittenNeedles
+			// The update only lands while the filer still holds the entry as
+			// the walk read it, so keep that copy before the first chunk is
+			// rewritten in place.
+			var readEntry *filer_pb.Entry
+			keepReadEntry := func() {
+				if readEntry == nil {
+					readEntry = proto.Clone(entry).(*filer_pb.Entry)
+				}
+			}
 			for i, chunk := range entry.Chunks {
 				if chunk.IsChunkManifest {
 					if !c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.VolumeId)) {
@@ -190,14 +210,18 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 					if !changed || !*apply {
 						continue
 					}
+					keepReadEntry()
 					entry.Chunks[i] = newChunk
 					entryChanged = true
-					movedSources = append(movedSources, rewritten.sources...)
+					moved.sources = append(moved.sources, rewritten.sources...)
 					// The old manifest needle is always orphaned when we
 					// replace it with a freshly uploaded one, even when the
 					// rewrite was triggered by sub-chunk moves rather than the
 					// manifest volume itself being in the plan.
-					movedSources = append(movedSources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid})
+					moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid})
+					moved.targets = append(moved.targets, rewritten.targets...)
+					moved.targets = append(moved.targets, orphanedNeedle{volumeId: newChunk.Fid.VolumeId, fileId: newChunk.GetFileIdString()})
+					moved.reservations = append(moved.reservations, rewritten.reservations...)
 					continue
 				}
 
@@ -206,6 +230,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 					continue
 				}
 				seen.record(chunkVolumeId)
+				keepReadEntry()
 
 				oldFid := chunk.GetFileIdString()
 				oldVid := chunk.Fid.VolumeId
@@ -224,20 +249,29 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 					continue
 				}
 				entryChanged = true
-				movedSources = append(movedSources, orphanedNeedle{volumeId: oldVid, fileId: oldFid})
+				moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldVid, fileId: oldFid})
+				moved.targets = append(moved.targets, orphanedNeedle{volumeId: uint32(toVolumeId), fileId: chunk.GetFileIdString()})
+				moved.reservations = append(moved.reservations, reservation{src: chunkVolumeId, dst: toVolumeId, size: chunk.Size})
 			}
 			if entryChanged {
-				if uErr := filer_pb.UpdateEntry(context.Background(), filerClient, &filer_pb.UpdateEntryRequest{
-					Directory: string(parentPath),
-					Entry:     entry,
-				}); uErr != nil {
+				stale, uErr := updateEntryIfUnchanged(context.Background(), filerClient, string(parentPath), entry, readEntry)
+				if stale {
+					fmt.Printf("skip %s: changed during merge\n", entryPath)
+					moved.targets = unreferencedTargets(context.Background(), filerClient, lookupFn, parentPath, entry.Name, moved.targets)
+					moved.abandon(commandEnv, entryPath, plan)
+					if len(entry.HardLinkId) > 0 {
+						processedHardLinks.Delete(string(entry.HardLinkId))
+					}
+					return nil
+				}
+				if uErr != nil {
 					fmt.Printf("failed to update %s: %v\n", entryPath, uErr)
 					// Filer still references the source fids. Deleting them
 					// now would lose data — abandon the cleanup for this
 					// entry and let fsck reconcile later.
 					return nil
 				}
-				deleteOrphanedNeedles(commandEnv, entryPath, movedSources, false)
+				deleteOrphanedNeedles(commandEnv, entryPath, moved.sources, false)
 			}
 			return nil
 		})
@@ -313,6 +347,61 @@ type rewrittenNeedles struct {
 	reservations []reservation
 }
 
+// abandon drops a rewrite the filer never took: the copies on the target
+// volumes belong to nobody, and the plan gets back the room they held, or a
+// multi-target merge keeps counting bytes it just deleted and skips later
+// chunks for no reason.
+func (r rewrittenNeedles) abandon(commandEnv *CommandEnv, entryPath util.FullPath, plan *mergePlan) {
+	deleteOrphanedNeedles(commandEnv, entryPath, r.targets, true)
+	for _, res := range r.reservations {
+		plan.release(res.src, res.dst, res.size)
+	}
+}
+
+// updateEntryIfUnchanged writes entry back only while the filer still holds
+// readEntry. An unconditional write would revert a change made since the read
+// and the filer would delete the newer chunks as garbage. stale reports that
+// the filer refused because the entry was changed or deleted, so nothing was
+// written and the copies made for this update are referenced by nothing.
+func updateEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, entry, readEntry *filer_pb.Entry) (stale bool, err error) {
+	err = filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{
+		Directory: dir,
+		Entry:     entry,
+		Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{
+			{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: readEntry},
+		}},
+	})
+	switch status.Code(err) {
+	case codes.FailedPrecondition, codes.NotFound:
+		return true, err
+	}
+	return false, err
+}
+
+// unreferencedTargets drops the copies the filer's current entry points at.
+// moveChunk keeps the key and cookie, so an overlapping run that moved the same
+// chunk to the same volume wrote the same fid, and its committed update is what
+// refused this one. When the entry cannot be read, no copy is reported.
+func unreferencedTargets(ctx context.Context, client filer_pb.SeaweedFilerClient, lookupFn wdclient.LookupFileIdFunctionType, dir util.FullPath, name string, targets []orphanedNeedle) []orphanedNeedle {
+	resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(dir), Name: name})
+	if errors.Is(err, filer_pb.ErrNotFound) {
+		return targets
+	}
+	if err != nil {
+		return nil
+	}
+	chunks := resp.Entry.GetChunks()
+	dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, lookupFn, chunks, 0, math.MaxInt64, nil)
+	if err != nil {
+		return nil
+	}
+	referenced := make(map[string]bool)
+	for _, chunk := range slices.Concat(chunks, dataChunks, manifestChunks) {
+		referenced[chunk.GetFileIdString()] = true
+	}
+	return slices.DeleteFunc(targets, func(t orphanedNeedle) bool { return referenced[t.fileId] })
+}
+
 // deleteOrphanedNeedles fans out BatchDelete RPCs to every replica of each
 // needle's volume. Errors are logged but never returned — the data is already
 // orphan at this point, so a failed cleanup just leaves work for a later fsck.
@@ -740,16 +829,8 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk(
 
 	var rewritten rewrittenNeedles
 	// Every path out of here that is not the successful return abandons the
-	// rewrite with the filer still pointing at the old manifest, so whatever
-	// already landed on the target volumes belongs to nobody. Give the plan
-	// back the room those copies were holding, or a multi-target merge keeps
-	// counting bytes it just deleted and skips later chunks for no reason.
-	abandon := func() {
-		deleteOrphanedNeedles(commandEnv, entryPath, rewritten.targets, true)
-		for _, r := range rewritten.reservations {
-			plan.release(r.src, r.dst, r.size)
-		}
-	}
+	// rewrite with the filer still pointing at the old manifest.
+	abandon := func() { rewritten.abandon(commandEnv, entryPath, plan) }
 	anySubChanged := false
 	for i, sub := range subChunks {
 		if sub.IsChunkManifest {

weed/shell/command_fs_merge_volumes_test.go

@@ -1,12 +1,22 @@
 package shell
 
 import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 func newMergeCmd(limitMB uint64, vols ...*master_pb.VolumeInformationMessage) *commandFsMergeVolumes {
@@ -394,3 +404,112 @@ func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) {
 		t.Fatalf("expected no warning when all needles were seen, got %q", sb.String())
 	}
 }
+
+type conditionalUpdateFiler struct {
+	filer_pb.SeaweedFilerClient
+	err       error
+	req       *filer_pb.UpdateEntryRequest
+	current   *filer_pb.Entry
+	lookupErr error
+	lookupReq *filer_pb.LookupDirectoryEntryRequest
+}
+
+func (s *conditionalUpdateFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	s.lookupReq = req
+	if s.lookupErr != nil {
+		return nil, s.lookupErr
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: s.current}, nil
+}
+
+func (s *conditionalUpdateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
+	s.req = req
+	if s.err != nil {
+		return nil, s.err
+	}
+	return &filer_pb.UpdateEntryResponse{}, nil
+}
+
+// The write-back is conditioned on the entry as it was read, and only a refusal
+// that proves nothing was written counts as stale: the caller then deletes the
+// copies it made, which would lose data if the update might have landed.
+func TestUpdateEntryIfUnchanged(t *testing.T) {
+	read := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}}
+	moved := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 7}}}}
+
+	for _, tc := range []struct {
+		name  string
+		err   error
+		stale bool
+	}{
+		{"committed", nil, false},
+		{"changed", status.Error(codes.FailedPrecondition, "precondition failed"), true},
+		{"deleted", status.Error(codes.NotFound, "not found"), true},
+		{"outcome unknown", status.Error(codes.Unavailable, "filer down"), false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			fake := &conditionalUpdateFiler{err: tc.err}
+			stale, err := updateEntryIfUnchanged(context.Background(), fake, "/dir", moved, read)
+			if stale != tc.stale || (err != nil) != (tc.err != nil) {
+				t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err)
+			}
+			clauses := fake.req.GetCondition().GetClauses()
+			if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL ||
+				!proto.Equal(clauses[0].ExpectedEntry, read) || !proto.Equal(fake.req.Entry, moved) {
+				t.Fatalf("request %v must write the moved entry conditioned on the read one", fake.req)
+			}
+		})
+	}
+}
+
+// A refused update may have been refused by an overlapping run that moved the
+// same chunk to the same volume: moveChunk keeps the key and cookie, so that
+// run's committed entry points at the very fid this run would delete.
+func TestUnreferencedTargets(t *testing.T) {
+	committed := orphanedNeedle{volumeId: 9, fileId: "9,0101000007"}
+	own := orphanedNeedle{volumeId: 9, fileId: "9,0202000007"}
+	manifestCopy := orphanedNeedle{volumeId: 9, fileId: "9,0303000007"}
+	targets := func() []orphanedNeedle { return []orphanedNeedle{committed, own, manifestCopy} }
+	direct := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId}}}
+
+	blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId, Size: 10}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write(blob)
+	}))
+	defer volumeServer.Close()
+	lookupFn := func(_ context.Context, fileId string) ([]string, error) {
+		if fileId != manifestCopy.fileId {
+			return nil, fmt.Errorf("volume of %s unavailable", fileId)
+		}
+		return []string{volumeServer.URL + "/" + fileId}, nil
+	}
+	viaManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: manifestCopy.fileId, IsChunkManifest: true, Size: 10}}}
+	unreadableManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: "9,0404000007", IsChunkManifest: true, Size: 10}}}
+
+	for _, tc := range []struct {
+		name      string
+		current   *filer_pb.Entry
+		lookupErr error
+		want      []orphanedNeedle
+	}{
+		{"another run committed one copy", direct, nil, []orphanedNeedle{own, manifestCopy}},
+		{"another run committed copies through a manifest", viaManifest, nil, []orphanedNeedle{own}},
+		{"manifest unreadable", unreadableManifest, nil, nil},
+		{"entry deleted", direct, status.Error(codes.NotFound, "not found"), targets()},
+		{"entry unreadable", direct, status.Error(codes.Unavailable, "filer down"), nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			fake := &conditionalUpdateFiler{current: tc.current, lookupErr: tc.lookupErr}
+			got := unreferencedTargets(context.Background(), fake, lookupFn, "/dir", "f", targets())
+			if !slices.Equal(got, tc.want) {
+				t.Fatalf("got %v, want %v", got, tc.want)
+			}
+			if fake.lookupReq.GetDirectory() != "/dir" || fake.lookupReq.GetName() != "f" {
+				t.Fatalf("looked up %v, want /dir f", fake.lookupReq)
+			}
+		})
+	}
+}

weed/shell/command_fs_meta_change_volume_id.go

@@ -12,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/protobuf/proto"
 )
 
 func init() {
@@ -78,6 +79,7 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
 		return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(*dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error {
 			if !entry.IsDirectory {
 				var hasChanges bool
+				var readEntry *filer_pb.Entry
 				for _, chunk := range entry.Chunks {
 					if chunk.IsChunkManifest {
 						fmt.Printf("Change volume id for large file is not implemented yet: %s/%s\n", parentPath, entry.Name)
@@ -85,6 +87,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
 					}
 					chunkVolumeId := chunk.Fid.VolumeId
 					if toVolumeId, found := mapping[needle.VolumeId(chunkVolumeId)]; found {
+						if readEntry == nil {
+							readEntry = proto.Clone(entry).(*filer_pb.Entry)
+						}
 						hasChanges = true
 						chunk.Fid.VolumeId = uint32(toVolumeId)
 						chunk.FileId = ""
@@ -93,10 +98,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
 				if hasChanges {
 					println("Updating", parentPath, entry.Name)
 					if *applyChanges {
-						if updateErr := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
-							Directory: string(parentPath),
-							Entry:     entry,
-						}); updateErr != nil {
+						if stale, updateErr := updateEntryIfUnchanged(context.Background(), client, string(parentPath), entry, readEntry); stale {
+							fmt.Printf("skip %s/%s: changed since it was read\n", parentPath, entry.Name)
+						} else if updateErr != nil {
 							fmt.Printf("failed to update %s/%s: %v\n", parentPath, entry.Name, updateErr)
 						}
 					}

weed/shell/command_s3_versions_audit.go

@@ -13,6 +13,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/protobuf/proto"
 )
 
 func init() {
@@ -228,6 +229,7 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient
 	if entry.Extended == nil {
 		return nil
 	}
+	readEntry := proto.Clone(entry).(*filer_pb.Entry)
 	delete(entry.Extended, s3_constants.ExtLatestVersionIdKey)
 	delete(entry.Extended, s3_constants.ExtLatestVersionFileNameKey)
 	// Also clear the cached list metadata so a stale size/mtime/etag
@@ -237,10 +239,9 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient
 	delete(entry.Extended, s3_constants.ExtLatestVersionETagKey)
 	delete(entry.Extended, s3_constants.ExtLatestVersionOwnerKey)
 	delete(entry.Extended, s3_constants.ExtLatestVersionIsDeleteMarker)
-	_, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
-		Directory: string(parentPath),
-		Entry:     entry,
-	})
+	// A version written since the audit read the directory has moved the
+	// pointer, and clearing it would hide that version.
+	_, err := updateEntryIfUnchanged(ctx, client, string(parentPath), entry, readEntry)
 	return err
 }
 

weed/shell/command_s3_versions_audit_test.go

@@ -0,0 +1,37 @@
+package shell
+
+import (
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+// The heal clears the pointer only while the directory still names the file
+// the audit found missing; a version written since then has moved it.
+func TestHealStrandedPointerIsConditional(t *testing.T) {
+	fake := &conditionalUpdateFiler{}
+	entry := &filer_pb.Entry{Name: "obj.versions", IsDirectory: true, Extended: map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey:       []byte("v1"),
+		s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"),
+		"keep":                                   []byte("x"),
+	}}
+	if err := healStrandedPointer(context.Background(), fake, "/buckets/b", entry); err != nil {
+		t.Fatalf("heal: %v", err)
+	}
+
+	clauses := fake.req.GetCondition().GetClauses()
+	if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL {
+		t.Fatalf("heal must be conditioned on the audited entry, got %v", fake.req.GetCondition())
+	}
+	if string(clauses[0].ExpectedEntry.Extended[s3_constants.ExtLatestVersionIdKey]) != "v1" {
+		t.Fatalf("expected entry must hold the pointer the audit saw, got %v", clauses[0].ExpectedEntry.Extended)
+	}
+	written := fake.req.Entry.Extended
+	_, idKept := written[s3_constants.ExtLatestVersionIdKey]
+	_, nameKept := written[s3_constants.ExtLatestVersionFileNameKey]
+	if idKept || nameKept || string(written["keep"]) != "x" {
+		t.Fatalf("written entry must clear only the pointer, got %v", written)
+	}
+}

weed/shell/command_volume_fsck.go

@@ -105,6 +105,16 @@ func (c *commandVolumeFsck) Help() string {
 	walked directory by directory; there, avoid running with -reallyDeleteFromVolume
 	while files are being renamed or moved.
 
+	With -findMissingChunksInFiler, -volumeId may name a volume the master no
+	longer lists, to find the filer entries that still reference it. A volume is
+	also unlisted while its server is offline, so purging these entries with
+	-reallyDeleteFilerEntries needs -confirmMissingVolumeDeletion as well; pass
+	it only after checking that every expected volume server is online. A purge
+	removes the whole filer entry, including its chunks on other volumes.
+	A -collection filter cannot be checked against an unlisted volume. With a
+	filter, a purge only reports references to unlisted volumes, and it refuses
+	to run when -volumeId names one.
+
 `
 }
 
@@ -123,6 +133,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
 	applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, "<expert only!> after detection, delete missing data from volumes / delete missing file entries from filer. Currently this only works with default filerGroup.")
 	c.forcePurging = fsckCommand.Bool("forcePurging", false, "delete missing data from volumes in one replica used together with applyPurging")
 	purgeAbsent := fsckCommand.Bool("reallyDeleteFilerEntries", false, "<expert only!> delete missing file entries from filer if the corresponding volume is missing for any reason, please ensure all still existing/expected volumes are connected! used together with findMissingChunksInFiler")
+	confirmMissingVolumeDeletion := fsckCommand.Bool("confirmMissingVolumeDeletion", false, "with -reallyDeleteFilerEntries, confirm that the -volumeId volumes the master does not list are retired")
 	c.purgeConcurrency = fsckCommand.Int("purgeConcurrency", 16, "parallel filer deletes for -reallyDeleteFilerEntries")
 	tempPath := fsckCommand.String("tempPath", path.Join(os.TempDir()), "path for temporary idx files")
 	cutoffTimeAgo := fsckCommand.Duration("cutoffTimeAgo", 5*time.Hour, "only include entries on volume servers before this cutoff time to check orphan chunks")
@@ -194,28 +205,26 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
 		}
 	}
 
-	// If the operator specified -volumeId, reject unknown ids up front instead
-	// of silently filtering them out and reporting "no orphan data". A silent
-	// success on a missing volume looks identical to a clean volume and hides
-	// typos, already-deleted volumes, and stale scripts.
 	if len(c.volumeIds) > 0 {
-		known := make(map[uint32]bool)
-		for _, vidMap := range dataNodeVolumeIdToVInfo {
-			for vid := range vidMap {
-				known[vid] = true
-			}
-		}
-		var missing []uint32
-		for vid := range c.volumeIds {
-			if !known[vid] {
-				missing = append(missing, vid)
-			}
+		missing, err := validateSelectedVolumeIDs(c.volumeIds, c.knownVolumeIds, *c.findMissingChunksInFiler)
+		if err != nil {
+			return err
 		}
 		if len(missing) > 0 {
-			sort.Slice(missing, func(i, j int) bool { return missing[i] < missing[j] })
-			return fmt.Errorf("volume(s) not found on master: %v", missing)
+			if *purgeAbsent {
+				if err := validateMissingVolumeCollectionFilter(missing, c.collectionMatcher); err != nil {
+					return err
+				}
+			}
+			if err := validateMissingVolumePurgeConfirmation(missing, *purgeAbsent, *confirmMissingVolumeDeletion); err != nil {
+				return err
+			}
+			fmt.Fprintf(c.writer, "WARNING: volume(s) not found on master (deleted or temporarily unavailable); scanning filer references for selected volume(s): %v\n", missing)
 		}
 	}
+	if *c.findMissingChunksInFiler && *purgeAbsent && c.collectionMatcher != nil {
+		fmt.Fprintln(c.writer, "WARNING: collection cannot be verified for volumes absent from master; Filer entries referencing those volumes will be reported but not purged.")
+	}
 
 	var collectCutoffFromAtNs int64 = 0
 	if cutoffTimeAgo.Seconds() != 0 {
@@ -360,13 +369,15 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 				// unknown), warn, and keep going. The unresolved counter blocks
 				// any purge step downstream so we never delete a sub-chunk we
 				// couldn't account for.
-				fmt.Fprintf(c.writer, "WARNING: ResolveChunkManifest failed for %s: %v — recording top-level chunk fids only; purging will be disabled\n",
+				fmt.Fprintf(c.writer, "WARNING: ResolveChunkManifest failed for %s: %v — recording top-level chunk fids only; orphan purging will be disabled\n",
 					util.NewFullPath(entry.Dir, entry.Entry.Name), resolveErr)
 				c.unresolvedManifestEntries.Add(1)
 				dataChunks = entry.Entry.GetChunks()
 				manifestChunks = nil
 			}
 			dataChunks = append(dataChunks, manifestChunks...)
+			// One Filer entry can have many missing chunks; delete the path only once.
+			purgePathQueued := false
 			for _, chunk := range dataChunks {
 				if cutoffFromAtNs != 0 && chunk.ModifiedTsNs > cutoffFromAtNs {
 					continue
@@ -374,6 +385,17 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 				if collectModifyFromAtNs != 0 && chunk.ModifiedTsNs < collectModifyFromAtNs {
 					continue
 				}
+				selectedVolume := len(c.volumeIds) == 0
+				if len(c.volumeIds) > 0 {
+					_, selectedVolume = c.volumeIds[chunk.Fid.VolumeId]
+				}
+				purgePath := purgeAbsent &&
+					selectedVolume &&
+					!c.knownVolumeIds[chunk.Fid.VolumeId] &&
+					!purgePathQueued && c.collectionMatcher == nil
+				if purgePath {
+					purgePathQueued = true
+				}
 				select {
 				case outputChan <- &Item{
 					vid:        chunk.Fid.VolumeId,
@@ -381,6 +403,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 					cookie:     chunk.Fid.Cookie,
 					path:       util.NewFullPath(entry.Dir, entry.Entry.Name),
 					hardLinkId: entry.Entry.GetHardLinkId(),
+					purgePath:  purgePath,
 				}:
 				case <-ctx.Done():
 					return ctx.Err()
@@ -422,7 +445,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 						continue
 					}
 					fmt.Fprintf(c.writer, "%d,%x%08x %s volume not found\n", i.vid, i.fileKey, i.cookie, i.path)
-					if purgeAbsent {
+					if purgeAbsent && i.purgePath {
 						fmt.Fprintf(c.writer, "deleting path %s after volume not found\n", i.path)
 						queuePurge(purgeItem{path: i.path, hardLinkId: i.hardLinkId})
 					}
@@ -644,6 +667,7 @@ type Item struct {
 	cookie     uint32
 	path       util.FullPath
 	hardLinkId []byte
+	purgePath  bool
 }
 
 // lockedWriter serializes concurrent writers — the traversal, the consumer,
@@ -814,7 +838,9 @@ func (c *commandVolumeFsck) httpDelete(path util.FullPath) {
 		fmt.Fprintln(c.writer, "delete response Headers : ", resp.Header)
 	}
 
-	if resp.StatusCode < http.StatusBadRequest {
+	if resp.StatusCode >= http.StatusMultipleChoices {
+		fmt.Fprintf(c.writer, "DELETE %s failed: %s\n", path, resp.Status)
+	} else {
 		dir, _ := path.DirAndName()
 		c.purgedDirsLock.Lock()
 		c.purgedDirs[util.FullPath(dir)] = struct{}{}
@@ -1094,3 +1120,40 @@ func writeToFile(bytes []byte, fileName string) error {
 	_, err = dst.Write(bytes)
 	return err
 }
+
+// validateSelectedVolumeIDs rejects IDs the master does not list, unless the
+// reverse filer scan is looking for references to exactly those volumes.
+func validateSelectedVolumeIDs(requested, known map[uint32]bool, allowMissing bool) ([]uint32, error) {
+	var missing []uint32
+	for vid := range requested {
+		if !known[vid] {
+			missing = append(missing, vid)
+		}
+	}
+	if len(missing) == 0 {
+		return nil, nil
+	}
+	sort.Slice(missing, func(i, j int) bool { return missing[i] < missing[j] })
+	if !allowMissing {
+		return nil, fmt.Errorf("volume(s) not found on master: %v", missing)
+	}
+	return missing, nil
+}
+
+// validateMissingVolumeCollectionFilter fails closed because a deleted volume
+// has no master metadata from which fsck can verify its collection.
+func validateMissingVolumeCollectionFilter(missing []uint32, matcher *wildcard.CollectionMatcher) error {
+	if len(missing) == 0 || matcher == nil {
+		return nil
+	}
+	return fmt.Errorf("cannot apply -collection=%q to missing volume(s) %v: their collection is unknown", matcher.String(), missing)
+}
+
+// validateMissingVolumePurgeConfirmation requires operators to distinguish
+// permanent retirement from a master that temporarily lost an offline server.
+func validateMissingVolumePurgeConfirmation(missing []uint32, purgeAbsent, confirmed bool) error {
+	if len(missing) == 0 || !purgeAbsent || confirmed {
+		return nil
+	}
+	return fmt.Errorf("refusing to purge Filer entries for volume(s) absent from master: %v; verify all expected volumes are connected and pass -confirmMissingVolumeDeletion", missing)
+}

weed/shell/command_volume_fsck_live_test.go

@@ -0,0 +1,360 @@
+//go:build integration
+
+package shell
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+// Opt-in: runs report-only and confirmed-purge scans against real services
+// after a volume is deleted, and checks that a stopped server blocks the purge.
+func TestVolumeFsckFindsReferenceAfterRealVolumeDeletion(t *testing.T) {
+	weed := buildWeedForIntegration(t)
+	base := t.TempDir()
+	for _, name := range []string{"logs", "master", "volume", "filer"} {
+		if err := os.MkdirAll(filepath.Join(base, name), 0o755); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	portPairs := freeIntegrationPortPairs(t, 3)
+	masterHTTP, masterGRPC := portPairs[0][0], portPairs[0][1]
+	volumeHTTP, volumeGRPC := portPairs[1][0], portPairs[1][1]
+	filerHTTP, filerGRPC := portPairs[2][0], portPairs[2][1]
+	masterAddress := fmt.Sprintf("127.0.0.1:%d", masterHTTP)
+
+	startIntegrationProcess(t, base, weed, "master",
+		"master", "-ip=127.0.0.1", "-ip.bind=127.0.0.1",
+		fmt.Sprintf("-port=%d", masterHTTP), fmt.Sprintf("-port.grpc=%d", masterGRPC),
+		"-mdir="+filepath.Join(base, "master"), "-defaultReplication=000", "-volumeSizeLimitMB=64")
+	waitIntegrationTCP(t, masterHTTP, filepath.Join(base, "logs", "master.log"))
+	waitIntegrationTCP(t, masterGRPC, filepath.Join(base, "logs", "master.log"))
+
+	volumeProcess := startIntegrationProcess(t, base, weed, "volume",
+		"volume", "-ip=127.0.0.1", "-ip.bind=127.0.0.1",
+		fmt.Sprintf("-port=%d", volumeHTTP), fmt.Sprintf("-port.grpc=%d", volumeGRPC),
+		"-master="+masterAddress, "-dir="+filepath.Join(base, "volume"), "-max=10")
+	waitIntegrationTCP(t, volumeHTTP, filepath.Join(base, "logs", "volume.log"))
+	waitIntegrationTCP(t, volumeGRPC, filepath.Join(base, "logs", "volume.log"))
+	waitForMasterVolumeRegistration(t, masterHTTP, volumeHTTP, filepath.Join(base, "logs", "volume.log"))
+
+	startIntegrationProcess(t, base, weed, "filer",
+		"filer", "-ip=127.0.0.1", "-ip.bind=127.0.0.1",
+		fmt.Sprintf("-port=%d", filerHTTP), fmt.Sprintf("-port.grpc=%d", filerGRPC),
+		"-master="+masterAddress, "-defaultReplicaPlacement=000",
+		"-defaultStoreDir="+filepath.Join(base, "filer"))
+	waitIntegrationTCP(t, filerHTTP, filepath.Join(base, "logs", "filer.log"))
+	waitIntegrationTCP(t, filerGRPC, filepath.Join(base, "logs", "filer.log"))
+
+	filerURL := fmt.Sprintf("http://127.0.0.1:%d", filerHTTP)
+	request, err := http.NewRequest(http.MethodPut, filerURL+"/fsck-probe.txt", strings.NewReader("live integration data"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatalf("upload probe through Filer: %v", err)
+	}
+	response.Body.Close()
+	if response.StatusCode != http.StatusCreated && response.StatusCode != http.StatusAccepted {
+		t.Fatalf("Filer PUT status = %d, want 201 or 202", response.StatusCode)
+	}
+
+	filerAddress := fmt.Sprintf("127.0.0.1:%d", filerHTTP)
+	meta := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /fsck-probe.txt\nunlock\n")
+	fidPattern := regexp.MustCompile(`"fid"\s*:\s*\{\s*"volumeId"\s*:\s*(\d+)`)
+	match := fidPattern.FindStringSubmatch(meta)
+	if len(match) != 2 {
+		t.Fatalf("could not read chunk volume ID from Filer metadata: %s", meta)
+	}
+	volumeID := match[1]
+
+	// Delete only the physical volume. The Filer entry intentionally remains
+	// so fsck must discover a dangling reference rather than a missing file.
+	deleteOutput := runIntegrationShell(t, weed, masterAddress, filerAddress,
+		fmt.Sprintf("lock\nvolume.delete -node=127.0.0.1:%d -volumeId=%s\nunlock\n", volumeHTTP, volumeID))
+	recentOutput := runIntegrationShell(t, weed, masterAddress, filerAddress,
+		fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -volumeId=%s\nunlock\n", volumeID))
+	if strings.Contains(recentOutput, "/fsck-probe.txt") {
+		t.Fatalf("default cutoff unexpectedly reported a just-written file: %q", recentOutput)
+	}
+
+	fsckOutput := runIntegrationShell(t, weed, masterAddress, filerAddress,
+		fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -volumeId=%s -cutoffTimeAgo=0\nunlock\n", volumeID))
+	if !strings.Contains(fsckOutput, "scanning filer references for selected volume(s)") {
+		t.Fatalf("fsck did not treat the deleted ID as a missing-volume target; delete=%q fsck=%q", deleteOutput, fsckOutput)
+	}
+	if !strings.Contains(fsckOutput, "/fsck-probe.txt") || !strings.Contains(fsckOutput, "volume not found") {
+		t.Fatalf("fsck did not report the stale Filer reference: %q", fsckOutput)
+	}
+
+	scopedPurgeOutput := runIntegrationShell(t, weed, masterAddress, filerAddress,
+		"lock\nvolume.fsck -findMissingChunksInFiler -reallyDeleteFilerEntries -collection=logs -cutoffTimeAgo=0\nunlock\n")
+	if !strings.Contains(scopedPurgeOutput, "collection cannot be verified for volumes absent from master") {
+		t.Fatalf("fsck did not warn that a missing volume cannot be collection-scoped: %q", scopedPurgeOutput)
+	}
+	if strings.Contains(scopedPurgeOutput, "deleting path /fsck-probe.txt after volume not found") {
+		t.Fatalf("collection-filtered fsck queued an unverified missing-volume path: %q", scopedPurgeOutput)
+	}
+
+	// Report-only fsck must not delete the stale Filer entry.
+	metaAfter := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /fsck-probe.txt\nunlock\n")
+	if !strings.Contains(metaAfter, "fsck-probe.txt") {
+		t.Fatalf("report-only fsck changed or removed the Filer entry: %q", metaAfter)
+	}
+	// An entry whose chunk manifest lived on the retired volume is unreadable too.
+	filerConn, err := grpc.NewClient(fmt.Sprintf("127.0.0.1:%d", filerGRPC), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("dial Filer gRPC: %v", err)
+	}
+	defer filerConn.Close()
+	filerClient := filer_pb.NewSeaweedFilerClient(filerConn)
+	volumeIDUint, err := strconv.ParseUint(volumeID, 10, 32)
+	if err != nil {
+		t.Fatalf("parse volume ID %q: %v", volumeID, err)
+	}
+	manifestChunk := &filer_pb.FileChunk{
+		Fid:             &filer_pb.FileId{VolumeId: uint32(volumeIDUint), FileKey: 987654321, Cookie: 1},
+		Size:            21,
+		IsChunkManifest: true,
+	}
+	manifestChunk.FileId = manifestChunk.GetFileIdString()
+	now := time.Now().Unix()
+	if err := filer_pb.CreateEntry(context.Background(), filerClient, &filer_pb.CreateEntryRequest{
+		Directory: "/",
+		Entry: &filer_pb.Entry{
+			Name:       "manifest-probe.txt",
+			Attributes: &filer_pb.FuseAttributes{FileSize: 21, FileMode: 0o644, Mtime: now, Crtime: now},
+			Chunks:     []*filer_pb.FileChunk{manifestChunk},
+		},
+	}); err != nil {
+		t.Fatalf("create unresolved-manifest probe: %v", err)
+	}
+
+	purgeOutput := runIntegrationShell(t, weed, masterAddress, filerAddress,
+		fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -reallyDeleteFilerEntries -confirmMissingVolumeDeletion -volumeId=%s -cutoffTimeAgo=0\nunlock\n", volumeID))
+	if !strings.Contains(purgeOutput, "ResolveChunkManifest failed") {
+		t.Fatalf("fsck did not surface the unresolved manifest: %q", purgeOutput)
+	}
+	if !strings.Contains(purgeOutput, "deleting path /manifest-probe.txt") {
+		t.Fatalf("confirmed purge did not queue the entry whose manifest was on the retired volume: %q", purgeOutput)
+	}
+	manifestAfterPurge, err := http.Get(filerURL + "/manifest-probe.txt")
+	if err != nil {
+		t.Fatalf("read manifest probe after confirmed purge: %v", err)
+	}
+	manifestAfterPurge.Body.Close()
+	if manifestAfterPurge.StatusCode != http.StatusNotFound {
+		t.Fatalf("confirmed purge manifest probe GET status = %d, want 404", manifestAfterPurge.StatusCode)
+	}
+
+	if !strings.Contains(purgeOutput, "deleting path /fsck-probe.txt after volume not found") {
+		t.Fatalf("confirmed purge did not queue the direct missing-volume entry: %q", purgeOutput)
+	}
+	probeAfterPurge, err := http.Get(filerURL + "/fsck-probe.txt")
+	if err != nil {
+		t.Fatalf("read Filer probe after confirmed purge: %v", err)
+	}
+	probeAfterPurge.Body.Close()
+	if probeAfterPurge.StatusCode != http.StatusNotFound {
+		t.Fatalf("confirmed purge Filer GET status = %d, want 404", probeAfterPurge.StatusCode)
+	}
+	// A volume still listed by the master but whose server is unreachable is
+	// not proof that its references are stale. Purge must fail before touching
+	// the Filer entry when the volume index cannot be read.
+	offlineRequest, err := http.NewRequest(http.MethodPut, filerURL+"/offline-probe.txt", strings.NewReader("offline safety probe"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	offlineResponse, err := http.DefaultClient.Do(offlineRequest)
+	if err != nil {
+		t.Fatalf("upload offline safety probe: %v", err)
+	}
+	offlineResponse.Body.Close()
+	if offlineResponse.StatusCode != http.StatusCreated && offlineResponse.StatusCode != http.StatusAccepted {
+		t.Fatalf("offline safety PUT status = %d", offlineResponse.StatusCode)
+	}
+	offlineMeta := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /offline-probe.txt\nunlock\n")
+	offlineMatch := fidPattern.FindStringSubmatch(offlineMeta)
+	if len(offlineMatch) != 2 {
+		t.Fatalf("could not read offline probe volume ID: %s", offlineMeta)
+	}
+	if err := volumeProcess.Process.Kill(); err != nil {
+		t.Fatalf("stop volume server: %v", err)
+	}
+	waitIntegrationTCPClosed(t, volumeHTTP)
+
+	shellCtx, cancelShell := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancelShell()
+	fsckCmd := exec.CommandContext(shellCtx, weed, "shell", "-master="+masterAddress, "-filer="+filerAddress)
+	fsckCmd.Stdin = strings.NewReader(fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -reallyDeleteFilerEntries -volumeId=%s -cutoffTimeAgo=0\nunlock\n", offlineMatch[1]))
+	fsckBytes, fsckErr := fsckCmd.CombinedOutput()
+	offlineFsckOutput := string(fsckBytes)
+	if fsckErr == nil {
+		t.Fatalf("fsck unexpectedly purged metadata for a volume server outage: %q", offlineFsckOutput)
+	}
+	if !strings.Contains(offlineFsckOutput, "confirmMissingVolumeDeletion") && !strings.Contains(offlineFsckOutput, "failed to collect file ids") {
+		t.Fatalf("fsck failed for an unexpected reason: err=%v output=%q", fsckErr, offlineFsckOutput)
+	}
+	offlineMetaAfter := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /offline-probe.txt\nunlock\n")
+	if !strings.Contains(offlineMetaAfter, "offline-probe.txt") {
+		t.Fatalf("offline-volume fsck removed the live Filer entry: %q", offlineMetaAfter)
+	}
+}
+
+func buildWeedForIntegration(t *testing.T) string {
+	t.Helper()
+	// Reuse the CI/integration binary when supplied; otherwise build this checkout.
+	if binary := os.Getenv("WEED_BINARY"); binary != "" {
+		if _, err := os.Stat(binary); err != nil {
+			t.Fatalf("WEED_BINARY=%q is unavailable: %v", binary, err)
+		}
+		return binary
+	}
+	_, source, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("resolve integration test source path")
+	}
+	root := filepath.Clean(filepath.Join(filepath.Dir(source), "..", ".."))
+	binary := filepath.Join(t.TempDir(), "weed")
+	cmd := exec.Command("go", "build", "-o", binary, "./weed")
+	cmd.Dir = root
+	if out, err := cmd.CombinedOutput(); err != nil {
+		t.Fatalf("build weed: %v\n%s", err, out)
+	}
+	return binary
+}
+
+// freeIntegrationPortPairs reserves HTTP ports with the CLI's conventional
+// HTTP+10000 gRPC mapping. The low range also keeps the derived ports valid
+// for weed shell's master/Filer client configuration.
+func freeIntegrationPortPairs(t *testing.T, count int) [][2]int {
+	t.Helper()
+	var listeners []net.Listener
+	defer func() {
+		for _, listener := range listeners {
+			listener.Close()
+		}
+	}()
+
+	pairs := make([][2]int, 0, count)
+	for httpPort := 20000; httpPort < 30000 && len(pairs) < count; httpPort++ {
+		grpcPort := httpPort + 10000
+		httpListener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", httpPort))
+		if err != nil {
+			continue
+		}
+		grpcListener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", grpcPort))
+		if err != nil {
+			httpListener.Close()
+			continue
+		}
+		listeners = append(listeners, httpListener, grpcListener)
+		pairs = append(pairs, [2]int{httpPort, grpcPort})
+	}
+	if len(pairs) != count {
+		t.Fatalf("reserved %d of %d required HTTP/gRPC port pairs", len(pairs), count)
+	}
+	return pairs
+}
+
+func startIntegrationProcess(t *testing.T, base, weed, name string, args ...string) *exec.Cmd {
+	t.Helper()
+	logFile, err := os.Create(filepath.Join(base, "logs", name+".log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmd := exec.Command(weed, args...)
+	cmd.Stdout = logFile
+	cmd.Stderr = logFile
+	if err := cmd.Start(); err != nil {
+		logFile.Close()
+		t.Fatalf("start %s: %v", name, err)
+	}
+	t.Cleanup(func() {
+		if cmd.Process != nil {
+			_ = cmd.Process.Kill()
+		}
+		_ = cmd.Wait()
+		_ = logFile.Close()
+	})
+	return cmd
+}
+
+// waitForMasterVolumeRegistration prevents racing a successful HTTP listener
+// start against the volume server's first heartbeat to the master.
+func waitForMasterVolumeRegistration(t *testing.T, masterPort, volumePort int, logPath string) {
+	t.Helper()
+	deadline := time.Now().Add(20 * time.Second)
+	wantAddress := fmt.Sprintf("127.0.0.1:%d", volumePort)
+	for time.Now().Before(deadline) {
+		response, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/dir/status", masterPort))
+		if err == nil {
+			body, readErr := io.ReadAll(response.Body)
+			response.Body.Close()
+			if readErr == nil && bytes.Contains(body, []byte(wantAddress)) {
+				return
+			}
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	log, _ := os.ReadFile(logPath)
+	t.Fatalf("master did not register volume server %s; log:\n%s", wantAddress, bytes.TrimSpace(log))
+}
+
+func waitIntegrationTCPClosed(t *testing.T, port int) {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 100*time.Millisecond)
+		if err != nil {
+			return
+		}
+		conn.Close()
+		time.Sleep(50 * time.Millisecond)
+	}
+	t.Fatalf("service still listens on port %d after process was stopped", port)
+}
+
+func waitIntegrationTCP(t *testing.T, port int, logPath string) {
+	t.Helper()
+	deadline := time.Now().Add(20 * time.Second)
+	for time.Now().Before(deadline) {
+		conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 200*time.Millisecond)
+		if err == nil {
+			conn.Close()
+			return
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	log, _ := os.ReadFile(logPath)
+	t.Fatalf("service did not listen on port %d; log:\n%s", port, bytes.TrimSpace(log))
+}
+
+func runIntegrationShell(t *testing.T, weed, master, filer, input string) string {
+	t.Helper()
+	cmd := exec.Command(weed, "shell", "-master="+master, "-filer="+filer)
+	cmd.Stdin = strings.NewReader(input)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("weed shell failed: %v\n%s", err, out)
+	}
+	return string(out)
+}

weed/shell/command_volume_fsck_missing_volume_test.go

@@ -0,0 +1,184 @@
+package shell
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
+)
+
+// A selected volume can already be absent from the master when fsck is used to
+// clean stale Filer references after physical volume deletion. Only that
+// explicitly selected missing ID should enter the purge queue.
+func TestVolumeFsckFindMissingFilerEntriesForDeletedSelectedVolume(t *testing.T) {
+	tempDir := t.TempDir()
+	chunk := func(volumeID uint32, fileKey uint64) *filer_pb.FileChunk {
+		return &filer_pb.FileChunk{
+			Fid:  &filer_pb.FileId{VolumeId: volumeID, FileKey: fileKey, Cookie: 1},
+			Size: 1,
+		}
+	}
+	stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{
+		"/":            {IsDirectory: true},
+		"/archive":     {IsDirectory: true},
+		"/archive/old": {Chunks: []*filer_pb.FileChunk{chunk(99, 1), chunk(99, 4), chunk(100, 2), chunk(7, 5)}},
+		"/live":        {Chunks: []*filer_pb.FileChunk{chunk(7, 3)}},
+	}}
+
+	verbose := false
+	findMissing := true
+	var output bytes.Buffer
+	c := &commandVolumeFsck{
+		tempFolder:               tempDir,
+		verbose:                  &verbose,
+		findMissingChunksInFiler: &findMissing,
+		writer:                   &output,
+		env:                      startStubFiler(t, stub),
+		volumeIds:                map[uint32]bool{99: true},
+		knownVolumeIds:           map[uint32]bool{7: true},
+	}
+	currentVolumes := map[string]map[uint32]VInfo{"dn1": {7: {collection: "live"}}}
+	var purged []purgeItem
+	if err := c.collectFilerFileIdAndPaths(currentVolumes, true, func(item purgeItem) {
+		purged = append(purged, item)
+	}, 0, 0); err != nil {
+		t.Fatalf("collect filer references: %v", err)
+	}
+
+	if len(purged) != 1 || purged[0].path != util.FullPath("/archive/old") {
+		t.Fatalf("purged paths = %v, want only /archive/old for selected missing volume 99", purged)
+	}
+	// The same logical file also references a live volume. The scan preserves
+	// that reference while queuing the path only once for the missing chunks.
+	foundSurvivingChunk := false
+	if err := c.readFilerFileIdFile(7, func(_ types.NeedleId, itemPath util.FullPath, _ []byte) {
+		if itemPath == util.FullPath("/archive/old") {
+			foundSurvivingChunk = true
+		}
+	}); err != nil {
+		t.Fatalf("read surviving volume references: %v", err)
+	}
+	if !foundSurvivingChunk {
+		t.Fatal("the surviving chunk reference was not recorded for volume 7")
+	}
+	for _, line := range []string{
+		"99,100000001 /archive/old volume not found\n",
+		"99,400000001 /archive/old volume not found\n",
+	} {
+		if !strings.Contains(output.String(), line) {
+			t.Errorf("missing-volume report %q lacks %q", output.String(), line)
+		}
+	}
+	if strings.Contains(output.String(), "100,200000001") {
+		t.Errorf("missing-volume report %q includes unselected volume 100", output.String())
+	}
+}
+
+// Unknown IDs remain errors outside the reverse Filer scan, preserving typo detection.
+func TestVolumeFsckSelectedVolumeIDValidation(t *testing.T) {
+	known := map[uint32]bool{7: true}
+	requested := map[uint32]bool{99: true, 100: true}
+
+	if _, err := validateSelectedVolumeIDs(requested, known, false); err == nil {
+		t.Fatal("normal fsck accepted volume IDs not present on the master")
+	}
+	missing, err := validateSelectedVolumeIDs(requested, known, true)
+	if err != nil {
+		t.Fatalf("reverse Filer scan rejected missing volume IDs: %v", err)
+	}
+	if len(missing) != 2 || missing[0] != 99 || missing[1] != 100 {
+		t.Fatalf("missing IDs = %v, want sorted [99 100]", missing)
+	}
+}
+
+// A missing volume has no trustworthy collection metadata; a collection
+// restriction must fail closed instead of silently widening or narrowing scope.
+func TestValidateMissingVolumeCollectionFilter(t *testing.T) {
+	missing := []uint32{12}
+	compile := func(filter string) *wildcard.CollectionMatcher {
+		matcher, err := wildcard.CompileCollectionMatcher(filter)
+		if err != nil {
+			t.Fatalf("compile %q: %v", filter, err)
+		}
+		return matcher
+	}
+	for _, filter := range []string{"", "*", "ALL_COLLECTIONS", "EACH_COLLECTION"} {
+		if err := validateMissingVolumeCollectionFilter(missing, compile(filter)); err != nil {
+			t.Errorf("filter %q: %v", filter, err)
+		}
+	}
+	if err := validateMissingVolumeCollectionFilter(missing, compile("photos")); err == nil || !strings.Contains(err.Error(), "collection is unknown") {
+		t.Fatalf("expected an unknown-collection error, got %v", err)
+	}
+}
+
+// Missing IDs require an extra acknowledgement only when metadata deletion
+// is requested; read-only diagnosis and known-volume purges are unchanged.
+func TestValidateMissingVolumePurgeConfirmation(t *testing.T) {
+	missing := []uint32{12}
+	for _, test := range []struct {
+		name        string
+		missing     []uint32
+		purgeAbsent bool
+		confirmed   bool
+		wantError   bool
+	}{
+		{name: "read-only scan", missing: missing},
+		{name: "purge known volumes only", purgeAbsent: true},
+		{name: "confirmed missing volume purge", missing: missing, purgeAbsent: true, confirmed: true},
+		{name: "unconfirmed missing volume purge", missing: missing, purgeAbsent: true, wantError: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			err := validateMissingVolumePurgeConfirmation(test.missing, test.purgeAbsent, test.confirmed)
+			if (err != nil) != test.wantError {
+				t.Fatalf("validateMissingVolumePurgeConfirmation() error = %v, wantError %v", err, test.wantError)
+			}
+			if err != nil && !strings.Contains(err.Error(), "confirmMissingVolumeDeletion") {
+				t.Fatalf("error %q does not name the required acknowledgement flag", err)
+			}
+		})
+	}
+}
+
+func TestVolumeFsckPurgesEntryWhoseManifestIsOnMissingVolume(t *testing.T) {
+	manifest := &filer_pb.FileChunk{
+		Fid:             &filer_pb.FileId{VolumeId: 99, FileKey: 1, Cookie: 1},
+		Size:            1,
+		IsChunkManifest: true,
+	}
+	stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{
+		"/":     {IsDirectory: true},
+		"/big":  {Chunks: []*filer_pb.FileChunk{manifest}},
+		"/live": {Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 2, Cookie: 1}, Size: 1}}},
+	}}
+
+	verbose := false
+	findMissing := true
+	var output bytes.Buffer
+	c := &commandVolumeFsck{
+		tempFolder:               t.TempDir(),
+		verbose:                  &verbose,
+		findMissingChunksInFiler: &findMissing,
+		writer:                   &output,
+		env:                      startStubFiler(t, stub),
+		volumeIds:                map[uint32]bool{99: true},
+		knownVolumeIds:           map[uint32]bool{7: true},
+	}
+	var purged []purgeItem
+	if err := c.collectFilerFileIdAndPaths(map[string]map[uint32]VInfo{"dn1": {7: {}}}, true, func(item purgeItem) {
+		purged = append(purged, item)
+	}, 0, 0); err != nil {
+		t.Fatalf("collect filer references: %v", err)
+	}
+
+	if !strings.Contains(output.String(), "ResolveChunkManifest failed for /big") {
+		t.Fatalf("manifest on the missing volume resolved unexpectedly: %q", output.String())
+	}
+	if len(purged) != 1 || purged[0].path != util.FullPath("/big") {
+		t.Fatalf("purged paths = %v, want /big, whose manifest is on missing volume 99", purged)
+	}
+}

weed/shell/command_volume_fsck_test.go

@@ -56,9 +56,10 @@ func TestVolumeFsckHttpDeleteRecordsParentDirectory(t *testing.T) {
 	defer filer.Close()
 
 	verbose := false
+	var output bytes.Buffer
 	c := &commandVolumeFsck{
 		verbose:    &verbose,
-		writer:     io.Discard,
+		writer:     &output,
 		purgedDirs: make(map[util.FullPath]struct{}),
 		env:        &CommandEnv{option: &ShellOptions{FilerAddress: pb.ServerAddress(strings.TrimPrefix(filer.URL, "http://"))}},
 	}
@@ -72,6 +73,9 @@ func TestVolumeFsckHttpDeleteRecordsParentDirectory(t *testing.T) {
 	if _, found := c.purgedDirs["/locked/dir"]; found {
 		t.Errorf("expected a rejected delete not to be recorded, got %v", c.purgedDirs)
 	}
+	if got := output.String(); got != "DELETE /locked/dir/file.txt failed: 403 Forbidden\n" {
+		t.Errorf("delete output = %q, want only the rejected path", got)
+	}
 }
 
 // stubFilerServer keeps a flat set of full paths and enforces the same delete

weed/storage/needle_map_sorted_file.go

@@ -127,6 +127,9 @@ func (m *SortedFileNeedleMap) Delete(key NeedleId, offset Offset) error {
 		return err
 	}
 	_, _, err = erasure_coding.SearchNeedleFromSortedIndex(f.file, m.dbFileSize, key, erasure_coding.MarkNeedleDeleted)
+	if err == nil {
+		m.logDelete(size)
+	}
 
 	return err
 }

weed/storage/needle_map_sorted_file_test.go

@@ -65,6 +65,21 @@ func TestSortedFileNeedleMap_DeleteAppendsTombstone(t *testing.T) {
 		t.Fatalf("Delete: %v", err)
 	}
 
+	if got, want := m.DeletedCount(), 1; got != want {
+		t.Fatalf("DeletedCount after delete: got %d, want %d", got, want)
+	}
+	if got, want := m.DeletedSize(), uint64(1024); got != want {
+		t.Fatalf("DeletedSize after delete: got %d, want %d", got, want)
+	}
+
+	// Idempotent delete must not double count deletions.
+	if err := m.Delete(deletedKey, deletedOff); err != nil {
+		t.Fatalf("idempotent Delete: %v", err)
+	}
+	if got, want := m.DeletedCount(), 1; got != want {
+		t.Fatalf("DeletedCount after idempotent delete: got %d, want %d", got, want)
+	}
+
 	if got, want := fileSize(t, idxPath), wantIdxSize+int64(NeedleMapEntrySize); got != want {
 		t.Fatalf("idx size after delete: got %d, want %d (front-overwrite regression)", got, want)
 	}

weed/storage/store_mark_readonly_can_delete_test.go

@@ -176,3 +176,44 @@ func TestLowDiskVolumeReportsCanDelete(t *testing.T) {
 	require.True(t, message.ReadOnly)
 	require.False(t, message.ReadOnlyCanDelete)
 }
+
+// After a restart a canDelete volume is served from .sdx; its deletes must
+// still move the counters volume.deleteEmpty and vacuum read, and land them
+// where the next reload would.
+func TestMarkVolumeReadonlyCanDelete_CountsDeletesAfterRestart(t *testing.T) {
+	dir := t.TempDir()
+	store := newSingleDirStore(t, dir)
+	const vid = needle.VolumeId(13)
+	require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0))
+
+	needles := []*needle.Needle{newRandomNeedle(1), newRandomNeedle(2), newRandomNeedle(3)}
+	for _, n := range needles {
+		_, err := store.WriteVolumeNeedle(vid, n, true, false)
+		require.NoError(t, err)
+	}
+	require.NoError(t, store.MarkVolumeReadonly(vid, true, true))
+	store.Close()
+
+	store2 := newSingleDirStore(t, dir)
+	v := store2.GetVolume(vid)
+	require.NotNil(t, v)
+	require.IsType(t, &SortedFileNeedleMap{}, v.nm)
+	for _, n := range needles {
+		_, err := store2.DeleteVolumeNeedle(vid, n)
+		require.NoError(t, err)
+	}
+	counters := func(v *Volume) [4]uint64 {
+		return [4]uint64{v.FileCount(), v.DeletedCount(), v.ContentSize(), v.DeletedSize()}
+	}
+	afterDeletes := counters(v)
+	require.Equal(t, uint64(len(needles)), afterDeletes[0])
+	require.Equal(t, afterDeletes[0], afterDeletes[1], "an emptied volume must report every file deleted")
+	require.Equal(t, afterDeletes[2], afterDeletes[3])
+	store2.Close()
+
+	store3 := newSingleDirStore(t, dir)
+	v3 := store3.GetVolume(vid)
+	require.NotNil(t, v3)
+	require.Equal(t, afterDeletes, counters(v3))
+	store3.Close()
+}