diff --git a/weed/filer/filer_delete_collection_test.go b/weed/filer/filer_delete_collection_test.go index 3300d8324c..f44d3a0076 100644 --- a/weed/filer/filer_delete_collection_test.go +++ b/weed/filer/filer_delete_collection_test.go @@ -125,6 +125,10 @@ func newFilerWithFakeMaster(t *testing.T) (*Filer, *hookedStore, *collectionDele LocalMetaLogBuffer: log_buffer.NewLogBuffer("test", time.Minute, func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, func() {}), } + t.Cleanup(func() { + f.LocalMetaLogBuffer.ShutdownLogBuffer() + f.LocalMetaLogBuffer.WaitForShutdown() + }) return f, store, master } diff --git a/weed/filer/filer_inode_test.go b/weed/filer/filer_inode_test.go index b00c1ec7e2..4d23f763bd 100644 --- a/weed/filer/filer_inode_test.go +++ b/weed/filer/filer_inode_test.go @@ -103,15 +103,26 @@ func TestEnsureEntryInodeHardLinkFitsSignedLong(t *testing.T) { } } -func newTestFilerWithStubStore() (*Filer, *stubFilerStore) { +func newTestFilerWithStubStore(t *testing.T) (*Filer, *stubFilerStore) { + t.Helper() store := newStubFilerStore() f := NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) f.Store = NewFilerStoreWrapper(store) + // A live LogBuffer keeps its flush loops and their multi-buffer pool + // reachable; without this the package's filers pin gigabytes on 32-bit. + // There is no master to flush to, so cancel the flush context first: + // otherwise the queued window retries forever and the buffer never exits. + t.Cleanup(func() { + close(f.deletionQuit) + f.flushCancel() + f.LocalMetaLogBuffer.ShutdownLogBuffer() + f.LocalMetaLogBuffer.WaitForShutdown() + }) return f, store } func TestCreateEntryAssignsInodeWhenMissing(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) entry := &Entry{ FullPath: util.FullPath("/dir/file.txt"), @@ -131,7 +142,7 @@ func TestCreateEntryAssignsInodeWhenMissing(t *testing.T) { } func TestCreateEntryAssignsInodesToAutoCreatedParents(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) entry := &Entry{ FullPath: util.FullPath("/a/b/c.txt"), @@ -152,7 +163,7 @@ func TestCreateEntryAssignsInodesToAutoCreatedParents(t *testing.T) { } func TestCreateEntryOExclPreservesExistingEntry(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) original := &Entry{ FullPath: util.FullPath("/buckets/my-bucket"), @@ -177,7 +188,7 @@ func TestCreateEntryOExclPreservesExistingEntry(t *testing.T) { } func TestCreateEntryOExclFailsOnLookupError(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) original := &Entry{ FullPath: util.FullPath("/buckets/my-bucket"), @@ -203,7 +214,7 @@ func TestCreateEntryOExclFailsOnLookupError(t *testing.T) { } func TestUpdateEntryPreservesExistingInode(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) original := &Entry{ FullPath: util.FullPath("/doc.txt"), @@ -240,7 +251,7 @@ func TestUpdateEntryPreservesExistingInode(t *testing.T) { } func TestUpdateEntryBackfillsMissingLegacyInode(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) original := &Entry{ FullPath: util.FullPath("/legacy.txt"), @@ -270,7 +281,7 @@ func TestUpdateEntryBackfillsMissingLegacyInode(t *testing.T) { // missing entry: on upsert stores proceeding would replace the entry without // loading the old chunks, orphaning them from cleanup. func TestCreateEntryFailsWhenLookupErrors(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) path := util.FullPath("/dir/file.txt") require.NoError(t, store.InsertEntry(context.Background(), &Entry{ FullPath: path, diff --git a/weed/filer/filer_lazy_remote_test.go b/weed/filer/filer_lazy_remote_test.go index 05d3105077..40d7c5358c 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -329,6 +329,12 @@ func newTestFiler(t *testing.T, store *stubFilerStore, rs *FilerRemoteStorage) * LocalMetaLogBuffer: log_buffer.NewLogBuffer("test", time.Minute, func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, func() {}), } + // A live LogBuffer keeps its flush loops and their multi-buffer pool + // reachable; without this the package's filers pin gigabytes on 32-bit. + t.Cleanup(func() { + f.LocalMetaLogBuffer.ShutdownLogBuffer() + f.LocalMetaLogBuffer.WaitForShutdown() + }) return f } diff --git a/weed/filer/filer_prefix_object_test.go b/weed/filer/filer_prefix_object_test.go index 42f9ba480c..3d399660f4 100644 --- a/weed/filer/filer_prefix_object_test.go +++ b/weed/filer/filer_prefix_object_test.go @@ -16,7 +16,7 @@ import ( // TestPromoteToPrefixObject covers a key written before the keys nested under it: // the file becomes the directory they live in, and has to stay an object of its own. func TestPromoteToPrefixObject(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) ctx := context.Background() object := &Entry{ @@ -49,7 +49,7 @@ func TestPromoteToPrefixObject(t *testing.T) { // TestExpiredDirectoryIsNotDeletedOnRead pins the other half: a TTL an older build // left on a promoted directory must not take the keys under it with it. func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) ctx := context.Background() dirPath := util.FullPath("/buckets/bkt/a/foo") @@ -80,7 +80,7 @@ func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) { // entries become invisible and their metadata row is removed without a // cache-specific transaction or reverse Volume index. func TestExpiredFileIsDeletedOnRead(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) ctx := context.Background() filePath := util.FullPath("/buckets/lmcache/expired") @@ -106,7 +106,7 @@ func TestExpiredFileIsDeletedOnRead(t *testing.T) { } func TestListDirectoryEntriesContextCanceled(t *testing.T) { - f, store := newTestFilerWithStubStore() + f, store := newTestFilerWithStubStore(t) ctx := context.Background() require.NoError(t, store.InsertEntry(ctx, &Entry{ diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 7b5d8c4a4f..96eedd0797 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -389,6 +389,13 @@ func (ma *MetaAggregator) Shutdown() { for { select { case <-done: + // The peers have stopped writing; retire the aggregated buffer so + // its flush loops and sealed windows are released with the + // aggregator instead of outliving it. + if ma.MetaLogBuffer != nil { + ma.MetaLogBuffer.ShutdownLogBuffer() + ma.MetaLogBuffer.WaitForShutdown() + } return case <-ticker.C: glog.Warningf("meta aggregator shutdown: peer subscriptions still stopping") diff --git a/weed/filer/meta_aggregator_test.go b/weed/filer/meta_aggregator_test.go index 82e8abd1a1..9ea75a4f68 100644 --- a/weed/filer/meta_aggregator_test.go +++ b/weed/filer/meta_aggregator_test.go @@ -11,6 +11,7 @@ import ( func TestOnPeerUpdateRepeatedAdd(t *testing.T) { peer := pb.ServerAddress("127.0.0.1:1") ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil) + defer ma.Shutdown() add := &master_pb.ClusterNodeUpdate{Address: string(peer), IsAdd: true} ma.OnPeerUpdate(add, time.Now()) diff --git a/weed/filer/meta_log_collection_test.go b/weed/filer/meta_log_collection_test.go index f5cef78783..b631a13ece 100644 --- a/weed/filer/meta_log_collection_test.go +++ b/weed/filer/meta_log_collection_test.go @@ -103,6 +103,12 @@ func TestViperReadsMetaLogOverrides(t *testing.T) { }() f := NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) + defer func() { + close(f.deletionQuit) + f.flushCancel() + f.LocalMetaLogBuffer.ShutdownLogBuffer() + f.LocalMetaLogBuffer.WaitForShutdown() + }() if got := f.metaLogTargetCollection; got != "filer-meta" { t.Fatalf("NewFiler did not read the collection override: %q", got) } diff --git a/weed/filer/metadata_event_sink_test.go b/weed/filer/metadata_event_sink_test.go index 80fac3248b..a41e28ddee 100644 --- a/weed/filer/metadata_event_sink_test.go +++ b/weed/filer/metadata_event_sink_test.go @@ -22,6 +22,10 @@ func TestNotifyUpdateEventRecordsRequestMetadataEvent(t *testing.T) { nil, ), } + defer func() { + f.LocalMetaLogBuffer.ShutdownLogBuffer() + f.LocalMetaLogBuffer.WaitForShutdown() + }() ctx, sink := WithMetadataEventSink(context.Background()) f.NotifyUpdateEvent(ctx, &Entry{FullPath: util.FullPath("/dir/file.txt")}, nil, true, false, []int32{7}) @@ -56,6 +60,10 @@ func TestNotifyUpdateEventReloadsLocalFilerConfiguration(t *testing.T) { nil, ), } + defer func() { + f.LocalMetaLogBuffer.ShutdownLogBuffer() + f.LocalMetaLogBuffer.WaitForShutdown() + }() updatedConf := NewFilerConf() if err := updatedConf.SetLocationConf(&filer_pb.FilerConf_PathConf{