diff --git a/weed/server/filer_grpc_server_chunk_condition_test.go b/weed/server/filer_grpc_server_chunk_condition_test.go index 6ebdcba65..7d2f75858 100644 --- a/weed/server/filer_grpc_server_chunk_condition_test.go +++ b/weed/server/filer_grpc_server_chunk_condition_test.go @@ -80,7 +80,7 @@ func TestUpdateEntryChunkConditionPreventsStrand(t *testing.T) { Attr: filer.Attr{Inode: 1, Mtime: time.Unix(1700000000, 0), Mode: 0644}, Chunks: storedChunks, } - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}, store } @@ -161,7 +161,7 @@ func TestDeleteEntryWaitsForPathLock(t *testing.T) { FullPath: "/test/obj", Attr: filer.Attr{Inode: 1, Mtime: time.Unix(1700000000, 0), Mode: 0644}, } - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) fs := &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()} lockPath := util.FullPath("/test/obj") diff --git a/weed/server/filer_grpc_server_condition_test.go b/weed/server/filer_grpc_server_condition_test.go index 2285a4bae..66906fafb 100644 --- a/weed/server/filer_grpc_server_condition_test.go +++ b/weed/server/filer_grpc_server_condition_test.go @@ -215,7 +215,7 @@ func TestCreateEntryConditionEnforced(t *testing.T) { Attr: filer.Attr{Inode: 1, Mtime: time.Unix(1700000000, 0)}, Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc")}, } - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) f.DirBucketsPath = "/buckets" fs := &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()} @@ -268,7 +268,7 @@ func TestCreateEntryReusesProvidedExisting(t *testing.T) { // during the overwrite are the same in both runs). store := newRenameTestStore() store.entries["/test/obj"] = existing.ShallowClone() - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) if err := f.CreateEntry(context.Background(), newEntry(), existing, false, false, nil, true, f.MaxFilenameLength); err != nil { t.Fatalf("create with existing: %v", err) } @@ -276,7 +276,7 @@ func TestCreateEntryReusesProvidedExisting(t *testing.T) { store2 := newRenameTestStore() store2.entries["/test/obj"] = existing.ShallowClone() - f2 := newRenameTestFiler(store2) + f2 := newRenameTestFiler(t, store2) if err := f2.CreateEntry(context.Background(), newEntry(), nil, false, false, nil, true, f2.MaxFilenameLength); err != nil { t.Fatalf("create with nil: %v", err) } @@ -298,7 +298,7 @@ func TestUpdateEntryConditionEnforced(t *testing.T) { Attr: filer.Attr{Inode: 1, Mtime: time.Unix(1700000000, 0), Mode: 0644}, Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc")}, } - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}, store } diff --git a/weed/server/filer_grpc_server_create_lock_test.go b/weed/server/filer_grpc_server_create_lock_test.go index ecd1f8ffb..30dc883fc 100644 --- a/weed/server/filer_grpc_server_create_lock_test.go +++ b/weed/server/filer_grpc_server_create_lock_test.go @@ -18,7 +18,7 @@ import ( func TestCreateEntryOExclSerialized(t *testing.T) { store := newRenameTestStore() store.findDelay = 5 * time.Millisecond - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) f.DirBucketsPath = "/buckets" fs := &FilerServer{ diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index 013bf7361..2aaa47e7a 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/weed/server/filer_grpc_server_object_txn_test.go @@ -17,13 +17,13 @@ import ( "google.golang.org/grpc/credentials/insecure" ) -func newTxnTestServer(seed map[string]*filer.Entry) (*FilerServer, *renameTestStore) { +func newTxnTestServer(t *testing.T, seed map[string]*filer.Entry) (*FilerServer, *renameTestStore) { store := newRenameTestStore() for path, entry := range seed { entry.FullPath = util.FullPath(path) store.entries[path] = entry } - f := newRenameTestFiler(store) + f := newRenameTestFiler(t, store) f.DirBucketsPath = "/buckets" fs := &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()} return fs, store @@ -34,7 +34,7 @@ func newTxnTestServer(seed map[string]*filer.Entry) (*FilerServer, *renameTestSt // all three atomically under one lock keyed on the object path. func TestObjectTransactionMultiEntry(t *testing.T) { now := time.Unix(1700000000, 0) - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj": { Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0644}, Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc")}, @@ -86,7 +86,7 @@ func TestObjectTransactionPatchNotifies(t *testing.T) { swapNotificationQueue(t, queue) now := time.Unix(1700000000, 0) - fs, _ := newTxnTestServer(map[string]*filer.Entry{ + fs, _ := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj/.versions": { Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755 | (1 << 31)}, Extended: map[string][]byte{"latest": []byte("v1")}, @@ -121,7 +121,7 @@ func TestObjectTransactionPatchNotifies(t *testing.T) { // and preserving the rest; without set_content, Content is left untouched. func TestObjectTransactionPatchContent(t *testing.T) { now := time.Unix(1700000000, 0) - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b": { Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0755 | (1 << 31)}, Extended: map[string][]byte{"versioning": []byte("Enabled")}, @@ -197,7 +197,7 @@ func TestObjectTransactionPatchContent(t *testing.T) { // A failing precondition aborts before any mutation is applied. func TestObjectTransactionPreconditionAborts(t *testing.T) { now := time.Unix(1700000000, 0) - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj": { Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0644}, Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc")}, @@ -236,7 +236,7 @@ func TestObjectTransactionRecomputeLatest(t *testing.T) { Extended: map[string][]byte{"vid": []byte(id), "etag": []byte("etag-" + id)}, } } - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj/.versions": { Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755 | (1 << 31)}, Extended: map[string][]byte{ @@ -304,7 +304,7 @@ func TestObjectTransactionRecomputeAscending(t *testing.T) { ver := func(id string) *filer.Entry { return &filer.Entry{Attr: filer.Attr{Inode: 10, Mtime: now, Crtime: now, Mode: 0644}, Extended: map[string][]byte{"vid": []byte(id)}} } - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj/.versions": {Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755 | (1 << 31)}, Extended: map[string][]byte{}}, "/buckets/b/obj/.versions/v1.ver": ver("v1"), "/buckets/b/obj/.versions/v2.ver": ver("v2"), @@ -339,7 +339,7 @@ func TestObjectTransactionBatchIndependent(t *testing.T) { Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc")}, } } - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/a": obj(1), "/buckets/b/c": obj(3), }) @@ -386,7 +386,7 @@ func TestObjectTransactionBatchIndependent(t *testing.T) { // panicking, keeping responses parallel to the requests. func TestObjectTransactionBatchNilTransaction(t *testing.T) { now := time.Unix(1700000000, 0) - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/a": {Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0644}}, }) @@ -418,7 +418,7 @@ func TestObjectTransactionBatchNilTransaction(t *testing.T) { // DELETE and PATCH of an absent entry are no-ops, so a replayed transaction // does not error. func TestObjectTransactionIdempotentNoops(t *testing.T) { - fs, _ := newTxnTestServer(nil) + fs, _ := newTxnTestServer(t, nil) req := &filer_pb.ObjectTransactionRequest{ LockKey: "/buckets/b/obj", @@ -449,7 +449,7 @@ func TestObjectTransactionRecomputeDemoteAndAttrs(t *testing.T) { Extended: map[string][]byte{"vid": []byte(id)}, } } - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj/.versions": { Attr: filer.Attr{Inode: 2, Mtime: t0, Crtime: t0, Mode: 0755 | (1 << 31)}, Extended: map[string][]byte{"latestName": []byte("v1.ver"), "latestVid": []byte("v1")}, @@ -503,7 +503,7 @@ func TestObjectTransactionRecomputeDemoteAndAttrs(t *testing.T) { func TestObjectTransactionPutThenRecomputeLatest(t *testing.T) { t0 := time.Unix(1700000000, 0) t1 := time.Unix(1700000100, 0) - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj/.versions": { Attr: filer.Attr{Inode: 2, Mtime: t0, Crtime: t0, Mode: 0755 | (1 << 31)}, Extended: map[string][]byte{"latestName": []byte("v1.ver"), "latestVid": []byte("v1")}, @@ -589,7 +589,7 @@ func TestObjectTransactionVersionDeleteWithWorm(t *testing.T) { } // Legal hold ON: the WORM guard blocks; version and pointer untouched. - fs, store := newTxnTestServer(seed(true)) + fs, store := newTxnTestServer(t, seed(true)) resp, err := fs.ObjectTransaction(context.Background(), mkReq()) if err != nil { t.Fatalf("err: %v", err) @@ -605,7 +605,7 @@ func TestObjectTransactionVersionDeleteWithWorm(t *testing.T) { } // No legal hold: pointer recomputes to v_b (excluding v_c), then v_c is deleted. - fs, store = newTxnTestServer(seed(false)) + fs, store = newTxnTestServer(t, seed(false)) resp, err = fs.ObjectTransaction(context.Background(), mkReq()) if err != nil || resp.Error != "" { t.Fatalf("unlocked delete failed: err=%v resp=%q", err, resp.Error) @@ -659,7 +659,7 @@ func TestObjectTransactionDeleteRemovesEmptyParent(t *testing.T) { } // Last version: the emptied directory goes with it. - fs, store := newTxnTestServer(seed(false)) + fs, store := newTxnTestServer(t, seed(false)) resp, err := fs.ObjectTransaction(context.Background(), mkReq()) if err != nil || resp.Error != "" { t.Fatalf("txn failed: err=%v resp=%q", err, resp.Error) @@ -672,7 +672,7 @@ func TestObjectTransactionDeleteRemovesEmptyParent(t *testing.T) { } // A remaining child keeps the directory (non-recursive teardown declines). - fs, store = newTxnTestServer(seed(true)) + fs, store = newTxnTestServer(t, seed(true)) resp, err = fs.ObjectTransaction(context.Background(), mkReq()) if err != nil || resp.Error != "" { t.Fatalf("txn failed: err=%v resp=%q", err, resp.Error) @@ -685,7 +685,7 @@ func TestObjectTransactionDeleteRemovesEmptyParent(t *testing.T) { } // Replay: the child is already gone, the empty parent is still tidied. - fs, store = newTxnTestServer(map[string]*filer.Entry{ + fs, store = newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj/.versions": { Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755 | (1 << 31)}, }, @@ -703,7 +703,7 @@ func TestObjectTransactionDeleteRemovesEmptyParent(t *testing.T) { // copy) while merging Extended. func TestObjectTransactionPatchTouchMtime(t *testing.T) { old := time.Unix(1600000000, 0) - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj": { FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mtime: old, Crtime: old, Mode: 0644}, @@ -748,7 +748,7 @@ func withRing(fs *FilerServer, self pb.ServerAddress, servers ...pb.ServerAddres // forwarding to itself. func TestObjectTransactionRouteKeyOwnerAppliesLocally(t *testing.T) { self := pb.ServerAddress("localhost:1") - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj": {FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mode: 0644}}, }) withRing(fs, self, self) @@ -776,7 +776,7 @@ func TestObjectTransactionRouteKeyOwnerAppliesLocally(t *testing.T) { func TestObjectTransactionIsMovedSkipsForward(t *testing.T) { self := pb.ServerAddress("localhost:1") other := pb.ServerAddress("localhost:2") - fs, store := newTxnTestServer(map[string]*filer.Entry{ + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj": {FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mode: 0644}}, }) withRing(fs, self, other) // ring owner is "other", not self @@ -803,7 +803,7 @@ func TestObjectTransactionIsMovedSkipsForward(t *testing.T) { // sender, so it would re-forward and fail to dial unless is_moved is set on the // forwarded request — making this also assert that one-hop bound over the wire. func TestObjectTransactionForwardsToOwner(t *testing.T) { - owner, ownerStore := newTxnTestServer(map[string]*filer.Entry{ + owner, ownerStore := newTxnTestServer(t, map[string]*filer.Entry{ "/buckets/b/obj": {FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mode: 0644}}, }) @@ -827,7 +827,7 @@ func TestObjectTransactionForwardsToOwner(t *testing.T) { go srv.Serve(lis) t.Cleanup(srv.Stop) - self, selfStore := newTxnTestServer(nil) + self, selfStore := newTxnTestServer(t, nil) withRing(self, sender, ownerAddr) // ring owner is the real owner; self forwards self.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) diff --git a/weed/server/filer_grpc_server_object_txn_ttl_test.go b/weed/server/filer_grpc_server_object_txn_ttl_test.go index bb62690bf..124505036 100644 --- a/weed/server/filer_grpc_server_object_txn_ttl_test.go +++ b/weed/server/filer_grpc_server_object_txn_ttl_test.go @@ -18,7 +18,7 @@ func mustSetObjectTxnStorageConf(t *testing.T, fs *FilerServer, conf *filer_pb.F } func TestObjectTransactionPutAppliesConfiguredTTLAndExpires(t *testing.T) { - fs, store := newTxnTestServer(nil) + fs, store := newTxnTestServer(t, nil) mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ LocationPrefix: "/buckets/video/", @@ -72,7 +72,7 @@ func TestObjectTransactionPutAppliesConfiguredTTLAndExpires(t *testing.T) { } func TestObjectTransactionPutPreservesExplicitTTL(t *testing.T) { - fs, store := newTxnTestServer(nil) + fs, store := newTxnTestServer(t, nil) mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ LocationPrefix: "/buckets/video/", @@ -117,7 +117,7 @@ func TestObjectTransactionPutPreservesExplicitTTL(t *testing.T) { } func TestObjectTransactionPutClearsTTLForRemoteEntry(t *testing.T) { - fs, store := newTxnTestServer(nil) + fs, store := newTxnTestServer(t, nil) mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ LocationPrefix: "/buckets/video/", @@ -163,7 +163,7 @@ func TestObjectTransactionPutClearsTTLForRemoteEntry(t *testing.T) { } func TestObjectTransactionPutHonorsMaxFileNameLengthRule(t *testing.T) { - fs, store := newTxnTestServer(nil) + fs, store := newTxnTestServer(t, nil) mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ LocationPrefix: "/buckets/video/", @@ -205,7 +205,7 @@ func TestObjectTransactionPutHonorsMaxFileNameLengthRule(t *testing.T) { func TestObjectTransactionPutFailsOnReadOnlyStorageConfig(t *testing.T) { // An explicit TTL must not bypass the read-only rule. for _, ttlSec := range []int32{0, 7200} { - fs, store := newTxnTestServer(nil) + fs, store := newTxnTestServer(t, nil) mustSetObjectTxnStorageConf(t, fs, &filer_pb.FilerConf_PathConf{ LocationPrefix: "/buckets/video/", diff --git a/weed/server/filer_grpc_server_posix_lock_test.go b/weed/server/filer_grpc_server_posix_lock_test.go index bb05056d7..253999402 100644 --- a/weed/server/filer_grpc_server_posix_lock_test.go +++ b/weed/server/filer_grpc_server_posix_lock_test.go @@ -13,8 +13,8 @@ import ( "google.golang.org/grpc/credentials/insecure" ) -func newPosixTestServer() *FilerServer { - fs, _ := newTxnTestServer(nil) +func newPosixTestServer(t *testing.T) *FilerServer { + fs, _ := newTxnTestServer(t, nil) fs.posixLocks = posixlock.NewManager() return fs } @@ -33,7 +33,7 @@ func posixOp(t *testing.T, fs *FilerServer, op filer_pb.PosixLockOp, lk *filer_p } func TestPosixLockGrantAndConflict(t *testing.T) { - fs := newPosixTestServer() + fs := newPosixTestServer(t) if r := posixOp(t, fs, filer_pb.PosixLockOp_TRY_LOCK, pbLock(0, 99, posixlock.Write, 1, 1, 7, false)); !r.Granted { t.Fatal("first lock should be granted") @@ -49,7 +49,7 @@ func TestPosixLockGrantAndConflict(t *testing.T) { } func TestPosixLockUnlockThenReacquire(t *testing.T) { - fs := newPosixTestServer() + fs := newPosixTestServer(t) posixOp(t, fs, filer_pb.PosixLockOp_TRY_LOCK, pbLock(0, 99, posixlock.Write, 1, 1, 7, false)) posixOp(t, fs, filer_pb.PosixLockOp_UNLOCK, pbLock(0, 99, posixlock.Unlock, 1, 1, 7, false)) if r := posixOp(t, fs, filer_pb.PosixLockOp_TRY_LOCK, pbLock(0, 99, posixlock.Write, 2, 1, 8, false)); !r.Granted { @@ -58,7 +58,7 @@ func TestPosixLockUnlockThenReacquire(t *testing.T) { } func TestPosixLockGetLk(t *testing.T) { - fs := newPosixTestServer() + fs := newPosixTestServer(t) posixOp(t, fs, filer_pb.PosixLockOp_TRY_LOCK, pbLock(10, 50, posixlock.Write, 1, 1, 7, false)) r := posixOp(t, fs, filer_pb.PosixLockOp_GET_LK, pbLock(30, 70, posixlock.Read, 2, 1, 8, false)) @@ -73,7 +73,7 @@ func TestPosixLockGetLk(t *testing.T) { } func TestPosixLockReleasePosixOwnerKeepsFlock(t *testing.T) { - fs := newPosixTestServer() + fs := newPosixTestServer(t) posixOp(t, fs, filer_pb.PosixLockOp_TRY_LOCK, pbLock(0, 99, posixlock.Write, 1, 1, 7, false)) posixOp(t, fs, filer_pb.PosixLockOp_TRY_LOCK, pbLock(0, 1<<63, posixlock.Write, 1, 1, 7, true)) @@ -89,7 +89,7 @@ func TestPosixLockReleasePosixOwnerKeepsFlock(t *testing.T) { } func TestPosixLockKeepAlive(t *testing.T) { - fs := newPosixTestServer() + fs := newPosixTestServer(t) resp, err := fs.PosixLock(context.Background(), &filer_pb.PosixLockRequest{ Key: "s3.fuse.lock:/x", Op: filer_pb.PosixLockOp_KEEP_ALIVE, Lock: pbLock(0, 0, posixlock.Unlock, 7, 0, 0, false), @@ -109,7 +109,7 @@ func TestPosixLockKeepAlive(t *testing.T) { // sender, so without is_moved on the forwarded hop it would re-forward and fail. func TestPosixLockForwardsToOwner(t *testing.T) { const key = "s3.fuse.lock:/x" - owner := newPosixTestServer() + owner := newPosixTestServer(t) lis, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { @@ -126,7 +126,7 @@ func TestPosixLockForwardsToOwner(t *testing.T) { go srv.Serve(lis) t.Cleanup(srv.Stop) - self := newPosixTestServer() + self := newPosixTestServer(t) withRing(self, sender, ownerAddr) self.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) @@ -151,7 +151,7 @@ func TestPosixLockForwardsToOwner(t *testing.T) { } func TestPosixLockWarmupDefersGrants(t *testing.T) { - fs := newPosixTestServer() + fs := newPosixTestServer(t) fs.posixLockReadyAt.Store(time.Now().UnixNano()) // warming up // A would-be grant is deferred (not granted, no conflict) so the client retries. diff --git a/weed/server/filer_grpc_server_recompute_test.go b/weed/server/filer_grpc_server_recompute_test.go index a6354e3ba..18ecc6848 100644 --- a/weed/server/filer_grpc_server_recompute_test.go +++ b/weed/server/filer_grpc_server_recompute_test.go @@ -21,7 +21,7 @@ func TestRecomputeLatestEmitsPointerUpdateEvent(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store)} + server := &FilerServer{filer: newRenameTestFiler(t, store)} m := &filer_pb.ObjectMutation{ Type: filer_pb.ObjectMutation_RECOMPUTE_LATEST, Directory: "/buckets/b", @@ -82,7 +82,7 @@ func TestRecomputeLatestDemoteEmitsEvent(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store)} + server := &FilerServer{filer: newRenameTestFiler(t, store)} m := &filer_pb.ObjectMutation{ Type: filer_pb.ObjectMutation_RECOMPUTE_LATEST, Directory: "/buckets/b", diff --git a/weed/server/filer_grpc_server_rename_test.go b/weed/server/filer_grpc_server_rename_test.go index 5c6e058ae..be5a4476a 100644 --- a/weed/server/filer_grpc_server_rename_test.go +++ b/weed/server/filer_grpc_server_rename_test.go @@ -223,7 +223,12 @@ func swapNotificationQueue(t *testing.T, q notification.MessageQueue) { }) } -func newRenameTestFiler(store *renameTestStore) *filer.Filer { +// newRenameTestFiler builds a filer whose meta log buffer is released when the +// test ends. A live LogBuffer holds PreviousBufferCount+1 buffers of BufferSize +// each, and its loop goroutines keep the whole filer reachable, so the tens of +// filers this package builds would otherwise pin gigabytes for the rest of the +// run - fatal on 32-bit, where that exhausts the address space. +func newRenameTestFiler(t *testing.T, store *renameTestStore) *filer.Filer { dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) masterClient := wdclient.NewMasterClient( dialOption, @@ -235,19 +240,22 @@ func newRenameTestFiler(store *renameTestStore) *filer.Filer { *pb.NewServiceDiscoveryFromMap(map[string]pb.ServerAddress{}), ) + logBuffer := log_buffer.NewLogBuffer( + "test", + time.Minute, + func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, + nil, + func() {}, + ) + t.Cleanup(logBuffer.ShutdownLogBuffer) + return &filer.Filer{ - Store: filer.NewFilerStoreWrapper(store), - MasterClient: masterClient, - FilerConf: filer.NewFilerConf(), - RemoteStorage: filer.NewFilerRemoteStorage(), - MaxFilenameLength: 255, - LocalMetaLogBuffer: log_buffer.NewLogBuffer( - "test", - time.Minute, - func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, - nil, - func() {}, - ), + Store: filer.NewFilerStoreWrapper(store), + MasterClient: masterClient, + FilerConf: filer.NewFilerConf(), + RemoteStorage: filer.NewFilerRemoteStorage(), + MaxFilenameLength: 255, + LocalMetaLogBuffer: logBuffer, } } @@ -284,7 +292,7 @@ func TestAtomicRenameEntryEmitsLogicalRenameEvent(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store), entryLockTable: util.NewLockTable[util.FullPath]()} + server := &FilerServer{filer: newRenameTestFiler(t, store), entryLockTable: util.NewLockTable[util.FullPath]()} _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ OldDirectory: "/", OldName: "src.txt", @@ -334,7 +342,7 @@ func TestAtomicRenameEntryOverwriteEmitsDeleteThenRename(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store), entryLockTable: util.NewLockTable[util.FullPath]()} + server := &FilerServer{filer: newRenameTestFiler(t, store), entryLockTable: util.NewLockTable[util.FullPath]()} _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ OldDirectory: "/", OldName: "src.txt", @@ -398,7 +406,7 @@ func TestAtomicRenameEntryDoesNotEmitEventOnDeleteFailure(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store), entryLockTable: util.NewLockTable[util.FullPath]()} + server := &FilerServer{filer: newRenameTestFiler(t, store), entryLockTable: util.NewLockTable[util.FullPath]()} _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ OldDirectory: "/", OldName: "src.txt", @@ -422,7 +430,7 @@ func TestAtomicRenameEntryDoesNotEmitEventOnCommitFailure(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store), entryLockTable: util.NewLockTable[util.FullPath]()} + server := &FilerServer{filer: newRenameTestFiler(t, store), entryLockTable: util.NewLockTable[util.FullPath]()} _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ OldDirectory: "/", OldName: "src.txt", @@ -447,7 +455,7 @@ func TestAtomicRenameEntrySkipsDescendantTargetLookups(t *testing.T) { queue := &captureQueue{} swapNotificationQueue(t, queue) - server := &FilerServer{filer: newRenameTestFiler(store), entryLockTable: util.NewLockTable[util.FullPath]()} + server := &FilerServer{filer: newRenameTestFiler(t, store), entryLockTable: util.NewLockTable[util.FullPath]()} _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ OldDirectory: "/", OldName: "srcdir", diff --git a/weed/server/filer_grpc_server_traverse_meta_test.go b/weed/server/filer_grpc_server_traverse_meta_test.go index 7aa9e3a2e..cb2095111 100644 --- a/weed/server/filer_grpc_server_traverse_meta_test.go +++ b/weed/server/filer_grpc_server_traverse_meta_test.go @@ -63,7 +63,7 @@ func TestTraverseBfsMetadata(t *testing.T) { store.entries[file] = newFileEntry(file, 2) } - server := &FilerServer{filer: newRenameTestFiler(store)} + server := &FilerServer{filer: newRenameTestFiler(t, store)} stream := &captureTraverseStream{ctx: context.Background()} err := server.TraverseBfsMetadata(&filer_pb.TraverseBfsMetadataRequest{ diff --git a/weed/server/filer_server_tus_jwt_test.go b/weed/server/filer_server_tus_jwt_test.go index 1d77e20b5..af02a202e 100644 --- a/weed/server/filer_server_tus_jwt_test.go +++ b/weed/server/filer_server_tus_jwt_test.go @@ -46,7 +46,7 @@ func newTusTestServer(t *testing.T, sessions map[string]string) (*FilerServer, * t.Helper() store := newRenameTestStore() fs := &FilerServer{ - filer: newRenameTestFiler(store), + filer: newRenameTestFiler(t, store), filerGuard: security.NewGuard(nil, tusTestWriteKey, 0, tusTestReadKey, 0), option: &FilerOption{TusBasePath: "/.tus"}, } diff --git a/weed/server/filer_subscribe_loop_test.go b/weed/server/filer_subscribe_loop_test.go index 54e37231c..acfe91132 100644 --- a/weed/server/filer_subscribe_loop_test.go +++ b/weed/server/filer_subscribe_loop_test.go @@ -176,6 +176,11 @@ type subscribeHarness struct { // stalls the metadata log flush" state the whole PR exists to handle. gateMu sync.Mutex flushGate chan struct{} + + // flushMu guards stopped against the flushes still in flight when the test + // ends, so none of them writes through a store the cleanup has closed. + flushMu sync.RWMutex + stopped bool } const testFilerIdSuffix = "0000abcd" @@ -221,7 +226,12 @@ func newSubscribeHarness(t *testing.T) *subscribeHarness { // layer the way logFlushFunc writes through real volumes. f.LocalMetaLogBuffer.ShutdownLogBuffer() f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", time.Minute, h.flushToStore, nil, nil) - t.Cleanup(f.LocalMetaLogBuffer.ShutdownLogBuffer) + // Shutting the buffer down is not enough: NewFiler's deletion loop keeps the + // filer - and so the replacement buffer's tens of megabytes - reachable for + // the rest of the run. Quiesce the flush path first, since Filer.Shutdown + // closes the store a flush still in flight would write through. + t.Cleanup(f.Shutdown) + t.Cleanup(h.stopFlushes) h.fs = &FilerServer{ filer: f, @@ -248,6 +258,15 @@ func (h *subscribeHarness) releaseFlushes() { } } +// stopFlushes lets go of any gated flush and waits for the ones in flight to +// finish, then refuses the rest, so the store stays untouched from here on. +func (h *subscribeHarness) stopFlushes() { + h.releaseFlushes() + h.flushMu.Lock() + defer h.flushMu.Unlock() + h.stopped = true +} + func (h *subscribeHarness) flushToStore(lb *log_buffer.LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { h.gateMu.Lock() gate := h.flushGate @@ -256,6 +275,12 @@ func (h *subscribeHarness) flushToStore(lb *log_buffer.LogBuffer, startTime, sto <-gate } + h.flushMu.RLock() + defer h.flushMu.RUnlock() + if h.stopped { + return + } + // The same file naming and append shape as logFlushFunc, against the fake // volumes: one chunk per flushed window, named for the window start minute. startTime, stopTime = startTime.UTC(), stopTime.UTC() diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index 0bc6f8e96..279c343b6 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -193,7 +193,8 @@ type LogBuffer struct { // Notified only when a flush lands, for readers that cannot act on an append flushSubscribers map[string]chan struct{} isStopping *atomic.Bool - shutdownCh chan struct{} // closed by ShutdownLogBuffer to wake blocked subscribers + shutdownCh chan struct{} // closed by ShutdownLogBuffer to wake blocked subscribers + loopsDone sync.WaitGroup // loopFlush and loopInterval signal exit isAllFlushed bool flushChan chan *dataToFlush flushBudget *flushBudget @@ -235,6 +236,7 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc }, } lb.lastFlushedOffset.Store(-1) // Nothing flushed to disk yet + lb.loopsDone.Add(2) go lb.loopFlush() go lb.loopInterval() return lb @@ -740,6 +742,7 @@ func (logBuffer *LogBuffer) queueFlush(d *dataToFlush) bool { } func (logBuffer *LogBuffer) loopFlush() { + defer logBuffer.loopsDone.Done() for d := range logBuffer.flushChan { if d == nil { break // shutdown sentinel @@ -777,10 +780,17 @@ func (logBuffer *LogBuffer) loopFlush() { } func (logBuffer *LogBuffer) loopInterval() { - for !logBuffer.IsStopping() { - time.Sleep(logBuffer.flushInterval) - if logBuffer.IsStopping() { + defer logBuffer.loopsDone.Done() + // Wake on shutdown instead of sleeping through the interval: a goroutine + // parked in time.Sleep keeps the buffer and its ~40MB of slabs reachable + // for up to flushInterval after ShutdownLogBuffer. + ticker := time.NewTicker(logBuffer.flushInterval) + defer ticker.Stop() + for { + select { + case <-logBuffer.shutdownCh: return + case <-ticker.C: } logBuffer.Lock() diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go index 64798819c..10ea00902 100644 --- a/weed/util/log_buffer/log_buffer_test.go +++ b/weed/util/log_buffer/log_buffer_test.go @@ -576,3 +576,24 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) { t.Logf("✓ SUCCESS: Message received after %d disk read attempts", finalDiskReadCount) } } + +// A buffer shut down long before its flush interval elapses must not keep its +// loops - and the tens of megabytes of slabs they reach - alive until the next +// tick would have fired. +func TestShutdownStopsGoroutinesPromptly(t *testing.T) { + lb := NewLogBuffer("shutdown", time.Hour, nil, nil, nil) + // let both loops start and park in their waits before shutting down + time.Sleep(100 * time.Millisecond) + lb.ShutdownLogBuffer() + + done := make(chan struct{}) + go func() { + lb.loopsDone.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("log buffer goroutines still running 5s after shutdown") + } +}