fix: clone protobuf snapshots without copying internal locks (#11968)

* fix: clone protobuf snapshots without copying internal locks

* fix: reuse cloned snapshot fields when filtering

weed/admin/plugin/plugin_scheduler.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -614,7 +615,7 @@ func deriveSchedulerAdminRuntime(
 	descriptor *plugin_pb.JobTypeDescriptor,
 ) *plugin_pb.AdminRuntimeConfig {
 	if cfg != nil && cfg.AdminRuntime != nil {
-		adminConfig := *cfg.AdminRuntime
+		adminConfig := proto.Clone(cfg.AdminRuntime).(*plugin_pb.AdminRuntimeConfig)
 		// Overlay descriptor defaults for any zero numeric fields. Persisted
 		// configs from older versions have no execution_timeout_seconds, and
 		// without this overlay the scheduler would fall back to the 90s
@@ -646,7 +647,7 @@ func deriveSchedulerAdminRuntime(
 				adminConfig.ExecutionTimeoutSeconds = defaults.ExecutionTimeoutSeconds
 			}
 		}
-		return &adminConfig
+		return adminConfig
 	}
 
 	if descriptor == nil || descriptor.AdminRuntimeDefaults == nil {

weed/admin/plugin/registry.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"google.golang.org/protobuf/proto"
 )
 
 const defaultWorkerStaleTimeout = 2 * time.Minute
@@ -495,36 +496,35 @@ func cloneJobTypeCapability(in *plugin_pb.JobTypeCapability) *plugin_pb.JobTypeC
 	if in == nil {
 		return nil
 	}
-	out := *in
-	return &out
+	return proto.Clone(in).(*plugin_pb.JobTypeCapability)
 }
 
 func cloneWorkerHeartbeat(in *plugin_pb.WorkerHeartbeat) *plugin_pb.WorkerHeartbeat {
 	if in == nil {
 		return nil
 	}
-	out := *in
+	out := proto.Clone(in).(*plugin_pb.WorkerHeartbeat)
 	if in.RunningWork != nil {
-		out.RunningWork = make([]*plugin_pb.RunningWork, 0, len(in.RunningWork))
-		for _, rw := range in.RunningWork {
-			if rw == nil {
-				continue
+		running := out.RunningWork[:0]
+		// Clone can turn nil messages into empty messages, so filter using
+		// their original positions while reusing the cloned entries.
+		for i, work := range in.RunningWork {
+			if work != nil {
+				running = append(running, out.RunningWork[i])
 			}
-			clone := *rw
-			out.RunningWork = append(out.RunningWork, &clone)
 		}
-	}
-	if in.QueuedJobsByType != nil {
-		out.QueuedJobsByType = make(map[string]int32, len(in.QueuedJobsByType))
-		for k, v := range in.QueuedJobsByType {
-			out.QueuedJobsByType[k] = v
+		clear(out.RunningWork[len(running):])
+		if running == nil {
+			running = []*plugin_pb.RunningWork{}
 		}
+		out.RunningWork = running
 	}
-	if in.Metadata != nil {
-		out.Metadata = make(map[string]string, len(in.Metadata))
-		for k, v := range in.Metadata {
-			out.Metadata[k] = v
-		}
+	// Preserve writable empty maps without rebuilding populated clones.
+	if in.QueuedJobsByType != nil && out.QueuedJobsByType == nil {
+		out.QueuedJobsByType = make(map[string]int32)
 	}
-	return &out
+	if in.Metadata != nil && out.Metadata == nil {
+		out.Metadata = make(map[string]string)
+	}
+	return out
 }

weed/admin/plugin/registry_snapshot_test.go

@@ -0,0 +1,65 @@
+package plugin
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+func TestCloneWorkerHeartbeatFiltersNilAndIsolatesState(t *testing.T) {
+	t.Parallel()
+	source := &plugin_pb.WorkerHeartbeat{
+		RunningWork:      []*plugin_pb.RunningWork{nil, {WorkId: "first"}, nil, {WorkId: "second"}, nil},
+		QueuedJobsByType: map[string]int32{"vacuum": 2},
+		Metadata:         map[string]string{"rack": "rack-a"},
+	}
+	got := cloneWorkerHeartbeat(source)
+	if len(got.RunningWork) != 2 || got.RunningWork[0].GetWorkId() != "first" || got.RunningWork[1].GetWorkId() != "second" {
+		t.Fatalf("unexpected filtered work: %v", got.RunningWork)
+	}
+	got.RunningWork[0].WorkId = "changed"
+	got.QueuedJobsByType["vacuum"] = 99
+	got.Metadata["rack"] = "changed"
+	if len(source.RunningWork) != 5 || source.RunningWork[0] != nil || source.RunningWork[1].WorkId != "first" {
+		t.Fatalf("cloning mutated source work: %v", source.RunningWork)
+	}
+	if source.QueuedJobsByType["vacuum"] != 2 || source.Metadata["rack"] != "rack-a" {
+		t.Fatal("cloned maps alias the source")
+	}
+}
+
+func TestCloneWorkerHeartbeatPreservesEmptyCollections(t *testing.T) {
+	t.Parallel()
+	if cloneWorkerHeartbeat(nil) != nil {
+		t.Fatal("nil heartbeat must stay nil")
+	}
+	for _, tc := range []struct {
+		name        string
+		source      *plugin_pb.WorkerHeartbeat
+		wantNilWork bool
+	}{
+		{"nil", &plugin_pb.WorkerHeartbeat{}, true},
+		{"empty", &plugin_pb.WorkerHeartbeat{
+			RunningWork:      []*plugin_pb.RunningWork{},
+			QueuedJobsByType: map[string]int32{},
+			Metadata:         map[string]string{},
+		}, false},
+		{"all nil work", &plugin_pb.WorkerHeartbeat{RunningWork: []*plugin_pb.RunningWork{nil, nil}}, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got := cloneWorkerHeartbeat(tc.source)
+			if len(got.RunningWork) != 0 || (got.RunningWork == nil) != tc.wantNilWork {
+				t.Fatalf("unexpected empty work slice: %#v", got.RunningWork)
+			}
+			if (got.QueuedJobsByType == nil) != (tc.source.QueuedJobsByType == nil) || (got.Metadata == nil) != (tc.source.Metadata == nil) {
+				t.Fatal("clone changed whether empty maps are writable")
+			}
+			if got.Metadata != nil {
+				got.Metadata["added"] = "value"
+				if len(tc.source.Metadata) != 0 {
+					t.Fatal("empty metadata map aliases source")
+				}
+			}
+		})
+	}
+}

weed/cluster/lock_ring_manager.go

@@ -7,6 +7,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/protobuf/proto"
 )
 
 const LockRingStabilizationInterval = 1 * time.Second
@@ -102,9 +103,7 @@ func (lrm *LockRingManager) GetLastUpdate(filerGroup FilerGroupName) *master_pb.
 	if !ok || update == nil {
 		return nil
 	}
-	cp := *update
-	cp.Servers = append([]string(nil), update.Servers...)
-	return &cp
+	return proto.Clone(update).(*master_pb.LockRingUpdate)
 }
 
 // scheduleBroadcast resets the stabilization timer. If another change arrives

weed/credential/credential_manager.go

@@ -15,6 +15,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
 )
 
 // FilerAddressSetter is an interface for credential stores that need a dynamic filer address
@@ -168,15 +169,16 @@ func (cm *CredentialManager) SaveConfiguration(ctx context.Context, config *iam_
 	staticNames := cm.staticNames
 	cm.staticMu.RUnlock()
 	if len(staticNames) > 0 {
-		var dynamicOnly []*iam_pb.Identity
-		for _, ident := range config.Identities {
+		configCopy := proto.Clone(config).(*iam_pb.S3ApiConfiguration)
+		dynamicOnly := configCopy.Identities[:0]
+		for _, ident := range configCopy.Identities {
 			if !staticNames[ident.Name] {
 				dynamicOnly = append(dynamicOnly, ident)
 			}
 		}
-		configCopy := *config
+		clear(configCopy.Identities[len(dynamicOnly):])
 		configCopy.Identities = dynamicOnly
-		return cm.Store.SaveConfiguration(ctx, &configCopy)
+		return cm.Store.SaveConfiguration(ctx, configCopy)
 	}
 	return cm.Store.SaveConfiguration(ctx, config)
 }

weed/credential/credential_manager_snapshot_test.go

@@ -0,0 +1,46 @@
+package credential
+
+import (
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/protobuf/proto"
+)
+
+type snapshotConfigurationStore struct {
+	CredentialStore
+	saved *iam_pb.S3ApiConfiguration
+}
+
+func (s *snapshotConfigurationStore) SaveConfiguration(_ context.Context, config *iam_pb.S3ApiConfiguration) error {
+	s.saved = config
+	return nil
+}
+
+func TestSaveConfigurationFiltersStaticIdentitySnapshot(t *testing.T) {
+	t.Parallel()
+	store := &snapshotConfigurationStore{}
+	manager := &CredentialManager{Store: store}
+	manager.SetStaticIdentities([]*iam_pb.Identity{{Name: "static"}})
+	source := &iam_pb.S3ApiConfiguration{Identities: []*iam_pb.Identity{
+		{Name: "alice", Actions: []string{"Read"}},
+		{Name: "static"},
+		{Name: "bob", Actions: []string{"Write"}},
+	}}
+	before := proto.Clone(source)
+	if err := manager.SaveConfiguration(context.Background(), source); err != nil {
+		t.Fatal(err)
+	}
+	if store.saved == nil || len(store.saved.Identities) != 2 || store.saved.Identities[0].Name != "alice" || store.saved.Identities[1].Name != "bob" {
+		t.Fatalf("unexpected saved configuration: %v", store.saved)
+	}
+	if !proto.Equal(source, before) {
+		t.Fatal("filtering mutated the caller's configuration")
+	}
+	store.saved.Identities[0].Actions[0] = "Admin"
+	store.saved.Identities[1].Name = "changed"
+	if !proto.Equal(source, before) {
+		t.Fatal("saved identities alias the caller's configuration")
+	}
+}

weed/mount/peer_registrar_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
 )
 
 // fakeFilerClient captures MountRegister/MountList calls and lets the test
@@ -17,22 +18,21 @@ type fakeFilerClient struct {
 	filer_pb.SeaweedFilerClient // embed for methods we don't need
 
 	mu            sync.Mutex
-	registerCalls []filer_pb.MountRegisterRequest
+	registerCalls []*filer_pb.MountRegisterRequest
 	listResponse  filer_pb.MountListResponse
 }
 
 func (f *fakeFilerClient) MountRegister(ctx context.Context, req *filer_pb.MountRegisterRequest, opts ...grpc.CallOption) (*filer_pb.MountRegisterResponse, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	f.registerCalls = append(f.registerCalls, *req)
+	f.registerCalls = append(f.registerCalls, proto.Clone(req).(*filer_pb.MountRegisterRequest))
 	return &filer_pb.MountRegisterResponse{}, nil
 }
 
 func (f *fakeFilerClient) MountList(ctx context.Context, req *filer_pb.MountListRequest, opts ...grpc.CallOption) (*filer_pb.MountListResponse, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	resp := f.listResponse // value copy
-	return &resp, nil
+	return proto.Clone(&f.listResponse).(*filer_pb.MountListResponse), nil
 }
 
 // fakeFilerFleet maps each addr to its own fakeFilerClient so a test can

weed/plugin/worker/worker.go

@@ -660,8 +660,7 @@ func (w *Worker) buildHeartbeat() *plugin_pb.WorkerHeartbeat {
 		if work == nil {
 			continue
 		}
-		cloned := *work
-		running = append(running, &cloned)
+		running = append(running, proto.Clone(work).(*plugin_pb.RunningWork))
 	}
 	w.runningMu.RUnlock()
 

weed/replication/sink/filersink/fetch_write.go

@@ -229,8 +229,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk
 
 	manifestDataChunks := make([]*filer_pb.FileChunk, len(replicatedResolvedChunks))
 	for i, chunk := range replicatedResolvedChunks {
-		copied := *chunk
-		manifestDataChunks[i] = &copied
+		manifestDataChunks[i] = proto.Clone(chunk).(*filer_pb.FileChunk)
 	}
 	filer_pb.BeforeEntrySerialization(manifestDataChunks)
 	manifestData, err := proto.Marshal(&filer_pb.FileChunkManifest{

weed/s3api/chunked_reader_v4_test.go

@@ -111,7 +111,7 @@ func TestNewSignV4ChunkedReaderStreamingUnsignedPayloadTrailer(t *testing.T) {
 	runWithRequest(iam, req, t, expectedPayload)
 }
 
-func runWithRequest(iam IdentityAccessManagement, req *http.Request, t *testing.T, expectedPayload string) {
+func runWithRequest(iam *IdentityAccessManagement, req *http.Request, t *testing.T, expectedPayload string) {
 	reader, errCode := iam.newChunkedReader(req)
 	assert.NotNil(t, reader)
 	assert.Equal(t, s3err.ErrNone, errCode)
@@ -124,11 +124,11 @@ func runWithRequest(iam IdentityAccessManagement, req *http.Request, t *testing.
 	assert.Equal(t, expectedPayload, string(data))
 }
 
-func setupIam() IdentityAccessManagement {
+func setupIam() *IdentityAccessManagement {
 	// Create an IdentityAccessManagement instance
 	// Add default access keys and secrets
 
-	iam := IdentityAccessManagement{
+	iam := &IdentityAccessManagement{
 		identities:        []*Identity{},
 		accessKeyIdent:    map[string]*Identity{},
 		accounts:          map[string]*Account{},

weed/stats/disk.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"google.golang.org/protobuf/proto"
 )
 
 type DiskIOProbeConfig struct {
@@ -91,7 +92,7 @@ type diskState struct {
 	statFailureCount int
 	statSuccessCount int
 	lastStatErr      error
-	lastGoodStatus   volume_server_pb.DiskStatus
+	lastGoodStatus   *volume_server_pb.DiskStatus
 	hasLastGood      bool
 
 	samples []ioSample
@@ -497,7 +498,7 @@ func (s *diskState) updateErrorLocked(disk *volume_server_pb.DiskStatus, config
 }
 
 func (s *diskState) rememberGoodStatusLocked(disk *volume_server_pb.DiskStatus) {
-	s.lastGoodStatus = *disk
+	s.lastGoodStatus = proto.Clone(disk).(*volume_server_pb.DiskStatus)
 	s.hasLastGood = true
 }
 
@@ -508,7 +509,8 @@ func (s *diskState) applyLastGoodStatusLocked(disk *volume_server_pb.DiskStatus)
 
 	dir := disk.Dir
 	errorMessage := disk.Error
-	*disk = s.lastGoodStatus
+	proto.Reset(disk)
+	proto.Merge(disk, s.lastGoodStatus)
 	disk.Dir = dir
 	disk.Error = errorMessage
 }