fix #3231 prevent snapshot corruption

Signed-off-by: Sylvere Richard <sylvere.richard@gmail.com>

Author: nowheresly

manager/state/raft/transport/peer.go

@@ -145,8 +145,15 @@ func raftMessageStructSize(m *raftpb.Message) int {
 
 // Returns the max allowable payload based on MaxRaftMsgSize and
 // the struct size for the given raftpb.Message.
+// If the struct overhead exceeds GRPCMaxMsgSize (e.g. very large
+// snapshot metadata), the payload size is clamped to a minimum of
+// 64 KiB to guarantee forward progress when splitting.
 func raftMessagePayloadSize(m *raftpb.Message) int {
-	return GRPCMaxMsgSize - raftMessageStructSize(m)
+	s := GRPCMaxMsgSize - raftMessageStructSize(m)
+	if s < 64<<10 {
+		s = 64 << 10
+	}
+	return s
 }
 
 // Split a large raft message into smaller messages.
@@ -164,6 +171,11 @@ func splitSnapshotData(_ context.Context, m *raftpb.Message) []api.StreamRaftMes
 	// Get the max payload size.
 	payloadSize := raftMessagePayloadSize(m)
 
+	// Capture the full snapshot data before the loop, because
+	// we need to copy the Snapshot struct for each chunk to avoid
+	// mutating m.Snapshot.Data through the shared pointer.
+	fullData := m.Snapshot.Data
+
 	// split the snapshot into smaller messages.
 	for snapDataIndex := 0; snapDataIndex < size; {
 		chunkSize := size - snapDataIndex
@@ -172,9 +184,13 @@ func splitSnapshotData(_ context.Context, m *raftpb.Message) []api.StreamRaftMes
 		}
 
 		raftMsg := *m
+		// Copy the Snapshot to avoid mutating the original message's
+		// Snapshot.Data through the shared pointer.
+		snap := *m.Snapshot
+		raftMsg.Snapshot = &snap
 
 		// sub-slice for this snapshot chunk.
-		raftMsg.Snapshot.Data = m.Snapshot.Data[snapDataIndex : snapDataIndex+chunkSize]
+		raftMsg.Snapshot.Data = fullData[snapDataIndex : snapDataIndex+chunkSize]
 
 		snapDataIndex += chunkSize
 

manager/state/raft/transport/transport_test.go

@@ -105,6 +105,73 @@ func testSend(ctx context.Context, c *mockCluster, from uint64, to []uint64, msg
 	}
 }
 
+func TestSplitSnapshotData(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("NormalSnapshot", func(t *testing.T) {
+		m := newSnapshotMessage(1, 2)
+		msgs := splitSnapshotData(ctx, &m)
+		assert.NotEmpty(t, msgs)
+		// reassemble and verify
+		var assembled []byte
+		for _, msg := range msgs {
+			assembled = append(assembled, msg.Message.Snapshot.Data...)
+		}
+		assert.Equal(t, m.Snapshot.Data, assembled)
+	})
+
+	t.Run("LargeMetadataDoesNotPanic", func(t *testing.T) {
+		// Simulate a snapshot where struct overhead exceeds GRPCMaxMsgSize.
+		// This is the scenario from the bug: many cluster objects cause
+		// large raft message metadata, making payloadSize negative without
+		// the fix.
+		data := make([]byte, 5<<20) // 5 MiB snapshot data
+		for i := range data {
+			data[i] = byte(i % 256)
+		}
+		m := raftpb.Message{
+			Type: raftpb.MsgSnap,
+			From: 1,
+			To:   2,
+			Snapshot: &raftpb.Snapshot{
+				Data: data,
+				Metadata: raftpb.SnapshotMetadata{
+					Index: uint64(len(data)),
+					// Large ConfState to push struct size above GRPCMaxMsgSize
+					ConfState: raftpb.ConfState{
+						Voters: make([]uint64, 1<<20),
+					},
+				},
+			},
+		}
+		// Must not panic
+		msgs := splitSnapshotData(ctx, &m)
+		assert.NotEmpty(t, msgs)
+		// reassemble and verify
+		var assembled []byte
+		for _, msg := range msgs {
+			assembled = append(assembled, msg.Message.Snapshot.Data...)
+		}
+		assert.Equal(t, data, assembled)
+	})
+}
+
+func TestRaftMessagePayloadSizeMinimum(t *testing.T) {
+	// When struct overhead exceeds GRPCMaxMsgSize, payloadSize must
+	// be clamped to the minimum (64 KiB) instead of going negative.
+	// Use Entries to inflate the struct size well beyond GRPCMaxMsgSize.
+	bigEntry := raftpb.Entry{Data: make([]byte, GRPCMaxMsgSize)}
+	m := &raftpb.Message{
+		Type:    raftpb.MsgSnap,
+		Entries: []raftpb.Entry{bigEntry},
+		Snapshot: &raftpb.Snapshot{
+			Data: nil,
+		},
+	}
+	ps := raftMessagePayloadSize(m)
+	assert.Equal(t, 64<<10, ps, "payloadSize should be clamped to minimum 64 KiB")
+}
+
 func TestSend(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	c := newCluster()