From c2ca83c0bf2f0c13b8cb3398728a1d4fd3f669ce Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 2 Nov 2026 18:00:59 -0800 Subject: [PATCH 1/9] filer: say jwt.filer_signing.key gates only the IAM gRPC service (#12353) * filer: say jwt.filer_signing.key gates only the IAM gRPC service The key never covered the rest of the filer gRPC port, which can read and write the IAM store under /etc/iam. Point operators at [grpc.filer] mTLS for a port reachable beyond a trusted network instead of the key. Co-Authored-By: Claude Opus 5.5 * filer: note the IAM gRPC OIDC provider and role RPCs are refused without the key Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/command/filer.go | 9 +++------ weed/command/scaffold/security.toml | 8 +++++--- weed/server/filer_server_handlers_iam_grpc.go | 10 +++++----- 3 files changed, 13 insertions(+), 14 deletions(-) diff --git a/weed/command/filer.go b/weed/command/filer.go index 1cf033f55a..f791deacc8 100644 --- a/weed/command/filer.go +++ b/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) diff --git a/weed/command/scaffold/security.toml b/weed/command/scaffold/security.toml index f412765b2f..b347d818a2 100644 --- a/weed/command/scaffold/security.toml +++ b/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 = "" diff --git a/weed/server/filer_server_handlers_iam_grpc.go b/weed/server/filer_server_handlers_iam_grpc.go index be98356614..508dcd2c3b 100644 --- a/weed/server/filer_server_handlers_iam_grpc.go +++ b/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 From 199b2e2dd9342c4f38767d7271b2796f5f4be5c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20L=E1=BB=99c=20Ph=C3=BAc?= <112325531+omninguyen@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:08:21 +0700 Subject: [PATCH 2/9] fix(s3api): reject an invalid storage class before uploading the body (#12354) * fix(s3api): reject an invalid storage class before uploading the body * fix(s3api): reject an invalid storage class on CreateMultipartUpload * fix(s3api): validate storage class and tags before reading the PUT body * fix(s3api): validate storage class and tags with the other PUT headers Checking them next to ValidateRequestEncryption keeps the check ahead of any later hook that may answer the PUT without reaching putToFiler. Co-Authored-By: Claude Opus 5.5 * refactor(s3api): reuse validatePutHeaders in putToFiler Co-Authored-By: Claude Opus 5.5 * fix(s3api): store storage class and tags on directory-marker PUTs The marker path validated x-amz-storage-class and x-amz-tagging but its mkdir callback never wrote them, so a valid value was accepted and then dropped and GetObjectTagging on the marker came back empty. Store both the way putToFiler does. Co-Authored-By: Claude Opus 5.5 * test(s3api): read directory-marker tags back through GetObjectTagging Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3_metadata_util.go | 4 + weed/s3api/s3api_object_handlers_multipart.go | 5 + weed/s3api/s3api_object_handlers_put.go | 62 +++-- .../s3api_storage_class_validation_test.go | 262 ++++++++++++++++++ 4 files changed, 312 insertions(+), 21 deletions(-) create mode 100644 weed/s3api/s3api_storage_class_validation_test.go diff --git a/weed/s3api/s3_metadata_util.go b/weed/s3api/s3_metadata_util.go index 01e0adc194..fbbfec9425 100644 --- a/weed/s3api/s3_metadata_util.go +++ b/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) } diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go index 780041f5cf..ca39881a12 100644 --- a/weed/s3api/s3api_object_handlers_multipart.go +++ b/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) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 8e1f3b720c..f06360883d 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/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 diff --git a/weed/s3api/s3api_storage_class_validation_test.go b/weed/s3api/s3api_storage_class_validation_test.go new file mode 100644 index 0000000000..d6d51d1693 --- /dev/null +++ b/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(), ""+apiErr.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])) +} From af45ab8f2a26a00f16040b6b874124139a209e6b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 04:53:57 -0800 Subject: [PATCH 3/9] s3api: known-answer fixture generator for SSE-S3 and filer-encrypted chunks (#12355) * s3api: a known-answer fixture generator for SSE-S3 and filer-encrypted chunks TestGenerateDirectReadFixture writes, from this package's own encryption, the pieces a reader outside the gateway must handle: SSE-S3 chunks with their per-chunk metadata, the object-level key, a single-stream ciphertext, the KEK derived from s3.sse.key and wrapped under a passphrase in both on-disk formats, and chunks encrypted by the filer, plain and gzipped. It is skipped unless DIRECT_READ_FIXTURE names the output file. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * s3api: the fixture generator restores the key manager and checks the write The test set the global key manager's KEK and left it; a later test in the same run then found a signing key where it expects none. The previous key is restored with t.Cleanup. The fixture file's close error is checked too, so a delayed write failure fails the test instead of leaving a short fixture. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * s3api: the fixture generator unwraps both KEK formats before writing The legacy v1 wrap is built by hand in the test; reading v1 and v2 back through unwrapKEK ties both to the format the key manager accepts. Co-Authored-By: Claude Opus 5.5 * s3api: pin a legacy v1 wrapped KEK made outside the package Generating the v1 sample through legacyKEKWrappingSalt let a change to the salt pass on both sides; a vector wrapped elsewhere under the historical salt must still unwrap. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Fable 5.1 --- weed/s3api/direct_read_fixture_test.go | 159 ++++++++++++++++++++ weed/s3api/s3_sse_s3_kek_passphrase_test.go | 22 +++ 2 files changed, 181 insertions(+) create mode 100644 weed/s3api/direct_read_fixture_test.go diff --git a/weed/s3api/direct_read_fixture_test.go b/weed/s3api/direct_read_fixture_test.go new file mode 100644 index 0000000000..486432ca3b --- /dev/null +++ b/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) + } +} diff --git a/weed/s3api/s3_sse_s3_kek_passphrase_test.go b/weed/s3api/s3_sse_s3_kek_passphrase_test.go index ea9f25c0fa..9d930be951 100644 --- a/weed/s3api/s3_sse_s3_kek_passphrase_test.go +++ b/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) + } +} From e5bc658eb7b7763a8513abbd26e541fe938b990d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 04:56:15 -0800 Subject: [PATCH 4/9] test/s3vectors: a hardened cluster with TLS volume servers, read JWTs and encrypted chunks (#12356) * test/s3vectors: a hardened cluster, with TLS volume servers, read JWTs and encrypted chunks Three switches on the test cluster: VECTOR_TLS=1 puts the volume servers behind HTTPS requiring client certificates and enables https.client for every process; VECTOR_READ_JWT=1 gives every process a jwt.signing.read.key, so volume reads need a token; VECTOR_ENCRYPT_VOLUMES=1 starts the gateway with -encryptVolumeData. The integration matrix gains a hardened leg per volume server with all three on. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/s3vectors: the hardened cluster proves its enforcement Once the gateway is up, the harness checks that the hardening holds, not only that it is configured: a TLS volume server refuses plain HTTP (Go with HTTP 400, Rust with a TLS alert) and a connection without the client certificate, and answers with it; under a read key an unsigned read of a real chunk, written through the gateway and located through the filer, gets 401. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/s3vectors: the hardened cluster proves its chunks are encrypted With VECTOR_ENCRYPT_VOLUMES the probe reads the stored chunk straight from the volume server, signing the read when a read key is set, and requires a recorded cipher key and bytes that differ from the plaintext. Co-Authored-By: Claude Opus 5.5 * test/s3vectors: the signed probe read must return the chunk An empty 200 passed both branches; an unencrypted read now has to return the bytes written, and an encrypted one has to return something else. Co-Authored-By: Claude Opus 5.5 * test/s3vectors: a read key in the caller's environment does not leak into the cluster Only VECTOR_READ_JWT turns read JWTs on; an inherited WEED_JWT_SIGNING_READ_KEY made the volume servers refuse the probe's unsigned read in an encrypt-only run. Co-Authored-By: Claude Opus 5.5 * test/s3vectors: quote certificate paths as TOML strings in security.toml Co-Authored-By: Claude Opus 5.5 * test/s3vectors: keep non-ASCII certificate paths as UTF-8 in security.toml TOML has no surrogate-pair escapes, and the Rust volume server reads the quoted value without decoding escapes at all. Co-Authored-By: Claude Opus 5.5 * test/s3vectors: an encrypted probe chunk must not be a readable gzip stream The gateway may gzip the body before encrypting it, so a chunk uploaded compressed but unencrypted would differ from the body and still pass. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Fable 5.1 --- .github/workflows/rust-vector-tests.yml | 10 ++- test/s3vectors/conftest.py | 108 +++++++++++++++++++++++- 2 files changed, 114 insertions(+), 4 deletions(-) diff --git a/.github/workflows/rust-vector-tests.yml b/.github/workflows/rust-vector-tests.yml index 4bea1ed29c..ac6371a89a 100644 --- a/.github/workflows/rust-vector-tests.yml +++ b/.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 diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index 9c99a758e7..a127c34251 100644 --- a/test/s3vectors/conftest.py +++ b/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") From 222cfa8e2e0a441a91dc6dff04782dced44968f4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 06:31:15 -0800 Subject: [PATCH 5/9] admin: summarize the EC delete_count accounting warning per refresh (#12359) * admin: summarize the EC delete_count accounting warning per refresh collectCollectionStats runs on every admin collection, bucket and topology refresh, and it logged a Warning for each EC volume whose delete_count summed across shard holders exceeds its file_count. That sum is an expected accounting artefact (a tombstone can be recorded on more than one holder), it recurs identically on every pass, and on one cluster with hundreds of such volumes it grew the admin log to 20 GB, more than 99% of it this one line. Log one V(1) summary per pass with the affected volume count and keep the per-volume detail at V(2). The object count is still skipped for those volumes. * admin: test that an EC volume with deletes above file_count is skipped Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/admin/dash/admin_server.go | 12 +++++- .../dash/collect_collection_stats_test.go | 41 +++++++++++++++++++ 2 files changed, 51 insertions(+), 2 deletions(-) diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index fd9abe2e0d..27958dec6d 100644 --- a/weed/admin/dash/admin_server.go +++ b/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 } diff --git a/weed/admin/dash/collect_collection_stats_test.go b/weed/admin/dash/collect_collection_stats_test.go index 5451ba6369..2d26610e3e 100644 --- a/weed/admin/dash/collect_collection_stats_test.go +++ b/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, From e8b284fef5f380cbea9f8b733b411bedb5deb8e0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 06:32:49 -0800 Subject: [PATCH 6/9] admin: scope plugin job and activity fetches to the selected job type (#12360) * admin: scope plugin job and activity fetches to the selected job type The plugin page fetched the newest 500 jobs and 500 activities cluster-wide and filtered them by job type in the browser. On a busy cluster that window can hold none of the selected type (433 erasure_coding + 64 ec_balance jobs, 497 ec_balance activities), so /plugin/lanes/default/execution?job=ec_vacuum showed "No execution jobs tracked yet" while the server had seven recent ec_vacuum jobs. When a job-type tab is active, fetch the queue/detection/execution lists with job_type= so the server filters before applying the limit, and refetch when the selected job type changes. The unscoped lists are still fetched for the status cards and the overview per-type summary. * admin: drop a plugin monitor refresh whose job type is no longer selected A refresh started on one job-type tab could finish after the user moved to another tab and that tab's refresh had already landed. Its rows, all of the earlier job type, then replaced the current tab's and the renderers filtered them down to empty tables until the next refresh. Co-Authored-By: Claude Opus 5.5 * admin: show the lane-wide rows when a job-type scoped fetch fails On a job-type tab a failed scoped jobs or activities request threw before anything was stored, so the status cards, scheduler state and tables all stayed stale. Fall back to the unscoped rows, which the renderers already filter by job type. Co-Authored-By: Claude Opus 5.5 * admin: filter plugin jobs and activities by lane before the limit /api/plugin/jobs and /api/plugin/activities took the newest N rows across every lane and then dropped the other lanes' rows, so on a busy default lane a quieter lane's page could come back empty. Co-Authored-By: Claude Opus 5.5 * admin: keep a failed plugin monitor fetch's last rows and report it Falling back to the lane-wide rows when a scoped fetch failed could still blank the tables, since that window may hold none of the selected type, and the failure went unreported. A lane-wide or scheduler-state failure also threw before the scoped rows that did arrive were stored. Each request now keeps its previous value on failure: the scoped rows only while the tab is unchanged, otherwise the lane-wide rows. Whatever arrived is stored and rendered, then the first failure is rethrown so the caller reports it. Co-Authored-By: Claude Opus 5.5 * admin: report a monitor refresh failure apart from the action it follows expireJob and runWorkflow refreshed the monitor inside the action's try block, so a failed refresh after a successful expire or workflow run was reported as a failed action, and the expired job's detail was not reloaded. Catch the refresh separately, as runDetection already does. Co-Authored-By: Claude Opus 5.5 * admin: show a repeated plugin page refresh failure once refreshAll runs every 3 seconds and notify opens a modal, so a request that kept failing reopened the same alert on every tick. Report an error only when it differs from the last one, and forget it once a refresh succeeds. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/admin/dash/plugin_api.go | 25 +++++- weed/admin/dash/plugin_api_test.go | 60 ++++++++++++++ weed/admin/view/app/plugin.templ | 119 ++++++++++++++++++++-------- weed/admin/view/app/plugin_templ.go | 2 +- 4 files changed, 169 insertions(+), 37 deletions(-) diff --git a/weed/admin/dash/plugin_api.go b/weed/admin/dash/plugin_api.go index 93821b5c4d..58afa7e977 100644 --- a/weed/admin/dash/plugin_api.go +++ b/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. diff --git a/weed/admin/dash/plugin_api_test.go b/weed/admin/dash/plugin_api_test.go index bc74188aa3..a88e215a5b 100644 --- a/weed/admin/dash/plugin_api_test.go +++ b/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)) + } +} diff --git a/weed/admin/view/app/plugin.templ b/weed/admin/view/app/plugin.templ index 166e42106d..f0518c95bc 100644 --- a/weed/admin/view/app/plugin.templ +++ b/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) { '
Success: ' + escapeHtml(String(data.success_count || 0)) + '
' + '
Errors: ' + escapeHtml(String(data.error_count || 0)) + '
'; 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() { diff --git a/weed/admin/view/app/plugin_templ.go b/weed/admin/view/app/plugin_templ.go index fbc54d49a8..621a1031a8 100644 --- a/weed/admin/view/app/plugin_templ.go +++ b/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, "
Workers

0

Active Jobs

0

Activities (recent)

0

Next scheduled check

-

Per Job Type Summary
Job TypeActive JobsRecent Activities
Loading...
Scheduler State
Sequential scheduler with per-job runtime limits
Job TypeEnabledDetectorIn FlightMax RuntimeExec GlobalExec/WorkerExecutor WorkersEffective ExecLast Run
Loading...
Workers
WorkerAddressCapabilitiesLoad
Loading...
Job Type Configuration
Not loaded
Selected Job Type
-
Descriptor
Select a job type to load schema and config.
Admin Config Form
No admin form loaded.
Worker Config Form
No worker form loaded.
Job Scheduling Settings
How often to check for new work.
Per-attempt deadline for one job. Size-aware tasks may extend this automatically.
Next scheduled check
Scheduler
-
Not scheduled
Run History
Keep last 10 success + last 10 errors
Successful Runs
TimeJob IDWorkerDuration
No data
Error Runs
TimeJob IDWorkerError
No data
Detection Results
Run detection to see proposals.
Job Queue
States: pending/assigned/running
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Detection Jobs
Detection activities for selected job type
TimeJob TypeRequest IDWorkerStageSourceMessage
Loading...
Execution Jobs
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Execution Activities
Non-detection events only
TimeJob TypeJob IDSourceStageMessage
Loading...
Job Detail
Select a job to view details.
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 7, "
Workers

0

Active Jobs

0

Activities (recent)

0

Next scheduled check

-

Per Job Type Summary
Job TypeActive JobsRecent Activities
Loading...
Scheduler State
Sequential scheduler with per-job runtime limits
Job TypeEnabledDetectorIn FlightMax RuntimeExec GlobalExec/WorkerExecutor WorkersEffective ExecLast Run
Loading...
Workers
WorkerAddressCapabilitiesLoad
Loading...
Job Type Configuration
Not loaded
Selected Job Type
-
Descriptor
Select a job type to load schema and config.
Admin Config Form
No admin form loaded.
Worker Config Form
No worker form loaded.
Job Scheduling Settings
How often to check for new work.
Per-attempt deadline for one job. Size-aware tasks may extend this automatically.
Next scheduled check
Scheduler
-
Not scheduled
Run History
Keep last 10 success + last 10 errors
Successful Runs
TimeJob IDWorkerDuration
No data
Error Runs
TimeJob IDWorkerError
No data
Detection Results
Run detection to see proposals.
Job Queue
States: pending/assigned/running
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Detection Jobs
Detection activities for selected job type
TimeJob TypeRequest IDWorkerStageSourceMessage
Loading...
Execution Jobs
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Execution Activities
Non-detection events only
TimeJob TypeJob IDSourceStageMessage
Loading...
Job Detail
Select a job to view details.
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } From 76198d15cfcc6dbbc9015f9f16fafcc5ff16bdd8 Mon Sep 17 00:00:00 2001 From: Minsoo Choi <162654709+Neo1228@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:57:43 +0900 Subject: [PATCH 7/9] fix(fsck): inspect references to deleted volumes (#12357) * fix(fsck): inspect references to deleted volumes * volume.fsck: treat every match-all collection filter as global ALL_COLLECTIONS and EACH_COLLECTION compile to the same match-all matcher as an empty filter or "*", but the string check treated them as a restricted scope: a selected missing volume was rejected and references to master-absent volumes were reported without being purged. Use the compiled matcher instead. Co-Authored-By: Claude Opus 5.5 * volume.fsck: purge an entry whose manifest is on a missing volume A top-level chunk on a missing volume already proves the entry is unreadable, whether or not its manifests resolve. Skipping such entries kept the very case where the manifest needle sat on the retired volume, and the remaining unresolved manifest then blocked every later orphan purge. The missing-needle check on live volumes never had this gate, so both paths now treat these entries the same. The warning now says only orphan purging is disabled, which is the only thing the counter gates. Co-Authored-By: Claude Opus 5.5 * volume.fsck: report a failed Filer delete A rejected purge DELETE printed nothing unless -v was set, so the run looked successful after "deleting path" while the entry stayed. The filer answers 204 for an entry that is already gone, so only real failures print. Co-Authored-By: Claude Opus 5.5 * volume.fsck: tighten help text, comments and missing-volume tests The help now wraps like the rest of it and says that a purge removes the whole entry, including its chunks on other volumes. The live test header no longer calls the run report-only, and the reverse-scan test checks the report line for each missing chunk. Co-Authored-By: Claude Opus 5.5 * volume.fsck: drop the missing-volume design notes The notes said no destructive change was approved while the change adds a confirmed purge, still described the dropped manifest gate, and mostly recorded local runs and a reverse index that is not part of this change. The operator contract lives in help volume.fsck. Co-Authored-By: Claude Opus 5.5 * volume.fsck: allow a report-only missing-volume scan with -collection The collection check exists to stop a purge that cannot be scoped, but it also rejected a report-only run that deletes nothing. Apply it only with -reallyDeleteFilerEntries, and limit the -confirmMissingVolumeDeletion description to the -volumeId case it guards. Co-Authored-By: Claude Opus 5.5 * volume.fsck: count only a 2xx answer as a purged entry A redirect the client could not follow left the entry in place but was recorded as deleted, which also queued its parent for empty-directory cleanup. Co-Authored-By: Claude Opus 5.5 * volume.fsck: say that a filtered purge of a missing -volumeId is refused The help and the early warning both said a purge with -collection only reports references to unlisted volumes, but a -volumeId naming one is rejected before the scan. The help now says so, and the warning prints after that check, so a refused run no longer claims it will report. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: MZ01-CHOIMINSU Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/shell/command_volume_fsck.go | 103 ++++- weed/shell/command_volume_fsck_live_test.go | 360 ++++++++++++++++++ ...command_volume_fsck_missing_volume_test.go | 184 +++++++++ weed/shell/command_volume_fsck_test.go | 6 +- 4 files changed, 632 insertions(+), 21 deletions(-) create mode 100644 weed/shell/command_volume_fsck_live_test.go create mode 100644 weed/shell/command_volume_fsck_missing_volume_test.go diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 67b0fa08e5..efabf1ceb1 100644 --- a/weed/shell/command_volume_fsck.go +++ b/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, " 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, " 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) +} diff --git a/weed/shell/command_volume_fsck_live_test.go b/weed/shell/command_volume_fsck_live_test.go new file mode 100644 index 0000000000..15f35730ba --- /dev/null +++ b/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) +} diff --git a/weed/shell/command_volume_fsck_missing_volume_test.go b/weed/shell/command_volume_fsck_missing_volume_test.go new file mode 100644 index 0000000000..f19e28aa30 --- /dev/null +++ b/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) + } +} diff --git a/weed/shell/command_volume_fsck_test.go b/weed/shell/command_volume_fsck_test.go index 0c742f82ab..5993688436 100644 --- a/weed/shell/command_volume_fsck_test.go +++ b/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 From d13d975b6e66c01f4d00d9f8cfb82339c52d9daa Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 07:59:18 -0800 Subject: [PATCH 8/9] shell: keep fs.mergeVolumes and fs.distributeChunks from reverting entries written while they run (#12362) * shell: keep fs.mergeVolumes from reverting entries written during the merge The walk reads a page of entries and moves their chunks one at a time, then wrote each entry back unconditionally. A file overwritten in between got its old version back, and the filer deleted the new version's chunks as garbage, so an acknowledged write was lost without any error. The update now carries IF_ENTRY_EQUAL over the entry as the walk read it. When the filer refuses because the entry changed or was deleted, the copies made on the target volumes are deleted, their plan room is released, and the entry is skipped; the source needles stay where the filer still points. Co-Authored-By: Claude Opus 5.5 * shell: keep fs.distributeChunks from reverting a file overwritten while it runs The command looks the entry up once, moves its chunks, and then wrote that entry back unconditionally, so an overwrite during the redistribution was reverted and the filer deleted the new version's chunks as garbage. The update now carries IF_ENTRY_EQUAL over the looked-up entry. When the filer refuses because the file changed or was deleted, the moved copies and any new manifest chunks are deleted and the command reports it. Co-Authored-By: Claude Opus 5.5 * shell: keep s3.versions.audit -heal from clearing a pointer that just moved The heal cleared the latest-version pointer on the directory entry the walk read, unconditionally. A version written between the audit and the heal has already pointed the directory at a live file, and clearing that pointer hid it. The heal now carries IF_ENTRY_EQUAL over the audited entry and reports a refusal as a failed heal. Co-Authored-By: Claude Opus 5.5 * shell: keep fs.meta.changeVolumeId from reverting entries written meanwhile The walk rewrote each entry's volume ids and wrote back the copy it had read, unconditionally, so an overwrite between the read and the write was reverted and its chunks were deleted as garbage. The update now carries IF_ENTRY_EQUAL over the entry as read, and a refused entry is skipped and reported. Co-Authored-By: Claude Opus 5.5 * shell: keep the target copies another fs.mergeVolumes run committed moveChunk keeps the key and cookie, so two overlapping merges of the same source into the same target write the same fid. When one run commits, the other's update is refused, and deleting its copies deleted the needle the committed entry now points at. A refused update now re-reads the entry, resolving its manifests, and deletes only the copies it no longer references. If the entry cannot be read, the copies are left for volume.fsck. Co-Authored-By: Claude Opus 5.5 * shell: let a later hard-link sibling retry a merge the filer refused The first sibling of a HardLinkId claims the move for all of them. When its update is refused, the claim made every later sibling skip, so the shared chunks stayed on the source volume for the rest of the run. A refused update now releases the claim, and a sibling listed after the change retries the move against the entry as it is now. Co-Authored-By: Claude Opus 5.5 * shell: delete the manifest blobs a failed fold in fs.distributeChunks saved When re-manifestizing failed partway, the blobs already uploaded were only counted in a log line and stayed on their volumes with nothing referencing them. They have freshly assigned fids, so they are now deleted with the cookie-checked delete the other orphan cleanups use. Co-Authored-By: Claude Opus 5.5 * shell: check the heal clears the pointer file name too Co-Authored-By: Claude Opus 5.5 * shell: require the shell lock for fs.mergeVolumes -apply moveChunk keeps the key and cookie, so two overlapping runs that move the same chunk into the same volume write the same fid. Re-reading the entry before deleting a refused run's copies covers the run that already committed, but not one that commits after that read. With the lock, two applying runs can no longer overlap, as volume.fsck and volume.merge already require. A dry run still works without it. Co-Authored-By: Claude Opus 5.5 * shell: check unreferencedTargets reads the refused entry itself Co-Authored-By: Claude Opus 5.5 * shell: check unreferencedTargets keeps a copy a manifest references Co-Authored-By: Claude Opus 5.5 * shell: check unreferencedTargets keeps the manifest copy and skips an unreadable manifest Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/shell/command_fs_distribute_chunks.go | 37 +++++- weed/shell/command_fs_merge_volumes.go | 119 +++++++++++++++--- weed/shell/command_fs_merge_volumes_test.go | 119 ++++++++++++++++++ .../shell/command_fs_meta_change_volume_id.go | 12 +- weed/shell/command_s3_versions_audit.go | 9 +- weed/shell/command_s3_versions_audit_test.go | 37 ++++++ 6 files changed, 300 insertions(+), 33 deletions(-) create mode 100644 weed/shell/command_s3_versions_audit_test.go diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go index 17c6f4fdd9..bb05ce6217 100644 --- a/weed/shell/command_fs_distribute_chunks.go +++ b/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") diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 99b6140943..26059da18e 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/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 { diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index d02dd60b71..37500e2d45 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/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) + } + }) + } +} diff --git a/weed/shell/command_fs_meta_change_volume_id.go b/weed/shell/command_fs_meta_change_volume_id.go index ef5a9386a2..23bedd5f99 100644 --- a/weed/shell/command_fs_meta_change_volume_id.go +++ b/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) } } diff --git a/weed/shell/command_s3_versions_audit.go b/weed/shell/command_s3_versions_audit.go index 1be0b16ae1..c7b5aad164 100644 --- a/weed/shell/command_s3_versions_audit.go +++ b/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 } diff --git a/weed/shell/command_s3_versions_audit_test.go b/weed/shell/command_s3_versions_audit_test.go new file mode 100644 index 0000000000..b32c1051e4 --- /dev/null +++ b/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) + } +} From 5df7faf0e38cedb7ad1016969d140b1f02d3b203 Mon Sep 17 00:00:00 2001 From: Mohd Quamar Tyagi <104281681+Tyagiquamar@users.noreply.github.com> Date: Tue, 3 Nov 2026 21:35:45 +0530 Subject: [PATCH 9/9] fix(storage): update deletion metrics in SortedFileNeedleMap.Delete (#12363) (#12364) * fix(storage): update deletion metrics in SortedFileNeedleMap.Delete (#12363) * storage: test canDelete volume counters across restarts A canDelete volume reloaded from .sdx must count its deletes so volume.deleteEmpty and vacuum see it drain, and the counters it reports must match what the next reload computes. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/storage/needle_map_sorted_file.go | 3 ++ weed/storage/needle_map_sorted_file_test.go | 15 +++++++ .../store_mark_readonly_can_delete_test.go | 41 +++++++++++++++++++ 3 files changed, 59 insertions(+) diff --git a/weed/storage/needle_map_sorted_file.go b/weed/storage/needle_map_sorted_file.go index 5a30525213..949ca91598 100644 --- a/weed/storage/needle_map_sorted_file.go +++ b/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 } diff --git a/weed/storage/needle_map_sorted_file_test.go b/weed/storage/needle_map_sorted_file_test.go index c9c1cc496f..e628ae4647 100644 --- a/weed/storage/needle_map_sorted_file_test.go +++ b/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) } diff --git a/weed/storage/store_mark_readonly_can_delete_test.go b/weed/storage/store_mark_readonly_can_delete_test.go index 00376df8cf..40f536c732 100644 --- a/weed/storage/store_mark_readonly_can_delete_test.go +++ b/weed/storage/store_mark_readonly_can_delete_test.go @@ -152,3 +152,44 @@ func TestMountVolumeAnnouncesReadOnlyState(t *testing.T) { }) } } + +// 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() +}