generate vtproto marshalers for filer_pb and use them on the metadata log path (#10337)

* generate vtproto marshalers for filer_pb and use them on the metadata log path

Reflection-based proto.Unmarshal allocates a fresh message tree through
reflect.New on every call. On the metadata subscription fan-out the same
event is decoded once per subscriber, so reflect.New tops the decode
churn under many mounts.

Generate MarshalVT/UnmarshalVT/SizeVT for filer.proto (a separate
filer_vtproto.pb.go, filer.pb.go untouched) and call them on the log
entry marshal and the subscribe/replay decode paths. UnmarshalVT
allocates message structs directly and copies byte and string fields, so
it stays wire-compatible with proto.Unmarshal and preserves the
non-aliasing the persisted-log cache depends on.

For SubscribeMetadataResponse this cuts decode allocations 69 -> 50 and
~4.5us -> ~2.1us per event; the win scales with subscriber overlap.

* marshal log entries directly into the buffer

SizeVT is allocation-free and MarshalToSizedBufferVT writes into a
pre-sized slice, so the log entry can be marshaled straight into
logBuffer.buf. This drops the per-entry MarshalVT allocation and the
follow-up copy on the write path.

* expand vtproto benchmarks: marshal, decode, and marshal-into-buffer by chunk count

Parametrize by nested-message count (chunks per event) and add encode +
zero-alloc marshal-into-buffer benchmarks alongside the decode one, so
the write-path win from MarshalToSizedBufferVT is measurable too.

* keep proto.Unmarshal for metadata events to preserve UTF-8 validation

UnmarshalVT skips proto3's UTF-8 validation of string fields, so a
SubscribeMetadataResponse with an invalid-UTF-8 string (e.g. Directory
"\xff") that proto.Unmarshal rejects would decode and reach path
filtering and subscribers. Decode events with proto.Unmarshal again;
UnmarshalVT stays on the log entry paths, whose only variable-length
fields are bytes and so carry no UTF-8 constraint.

Tests cover the codec difference and that a malformed event is skipped
before delivery.

go.mod

@@ -442,7 +442,7 @@ require (
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
 	github.com/pkg/xattr v0.4.12 // indirect
-	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
+	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/putdotio/go-putio/putio v0.0.0-20200123120452-16d982cac2b8 // indirect
 	github.com/relvacode/iso8601 v1.7.0 // indirect

weed/filer/filer_notify_read.go

@@ -11,13 +11,11 @@ import (
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
-	"google.golang.org/protobuf/proto"
-
-	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 type LogFileEntry struct {
@@ -518,7 +516,7 @@ func (iter *LogFileIterator) getNext() (logEntry *filer_pb.LogEntry, err error)
 			return nil, fmt.Errorf("entry data %d bytes, expected %d bytes", n, size)
 		}
 		logEntry = &filer_pb.LogEntry{}
-		if err = proto.Unmarshal(entryData, logEntry); err != nil {
+		if err = logEntry.UnmarshalVT(entryData); err != nil {
 			return
 		}
 		if logEntry.TsNs <= iter.startTsNs {

weed/filer/persisted_log_cache.go

@@ -8,13 +8,11 @@ import (
 	"sync"
 	"time"
 
-	"golang.org/x/sync/semaphore"
-	"golang.org/x/sync/singleflight"
-	"google.golang.org/protobuf/proto"
-
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+	"golang.org/x/sync/semaphore"
+	"golang.org/x/sync/singleflight"
 )
 
 const (
@@ -201,10 +199,10 @@ func loadLogFileEntries(masterClient *wdclient.MasterClient, chunk *filer_pb.Fil
 // decodeLogRecords parses size-prefixed LogEntry records. A buffer that stops
 // mid-record, or whose size prefix is garbage (also the symptom of starting
 // mid-record), reports errLogChunkIncomplete with the cleanly decoded prefix.
-// Since proto.Unmarshal is permissive enough to accept misaligned bytes,
+// Since UnmarshalVT is permissive enough to accept misaligned bytes,
 // records must also satisfy the writer's invariants: never empty, a positive
 // timestamp, and strictly increasing within one flushed buffer.
-// proto.Unmarshal copies all bytes, so the entries do not alias data.
+// UnmarshalVT copies all bytes, so the entries do not alias data.
 func decodeLogRecords(data []byte) (entries []*filer_pb.LogEntry, cacheable bool, err error) {
 	var lastTsNs int64
 	for pos := 0; pos < len(data); {
@@ -220,7 +218,7 @@ func decodeLogRecords(data []byte) (entries []*filer_pb.LogEntry, cacheable bool
 			return entries, false, errLogChunkIncomplete
 		}
 		logEntry := &filer_pb.LogEntry{}
-		if unmarshalErr := proto.Unmarshal(data[pos+4:pos+4+size], logEntry); unmarshalErr != nil {
+		if unmarshalErr := logEntry.UnmarshalVT(data[pos+4 : pos+4+size]); unmarshalErr != nil {
 			return entries, false, errLogChunkIncomplete
 		}
 		if logEntry.TsNs <= lastTsNs {

weed/pb/Makefile

@@ -5,7 +5,7 @@ all: gen
 gen:
 	protoc master.proto --go_out=./master_pb --go-grpc_out=./master_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
 	protoc volume_server.proto --go_out=./volume_server_pb --go-grpc_out=./volume_server_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
-	protoc filer.proto --go_out=./filer_pb --go-grpc_out=./filer_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
+	protoc filer.proto --go_out=./filer_pb --go-grpc_out=./filer_pb --go-vtproto_out=./filer_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative --go-vtproto_opt=paths=source_relative,features=marshal+unmarshal+size
 	protoc remote.proto --go_out=./remote_pb --go-grpc_out=./remote_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
 	protoc iam.proto --go_out=./iam_pb --go-grpc_out=./iam_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
 	protoc mount.proto --go_out=./mount_pb --go-grpc_out=./mount_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative

weed/pb/filer_pb/filer_vtproto_test.go

@@ -0,0 +1,265 @@
+package filer_pb
+
+import (
+	"fmt"
+	"testing"
+
+	"google.golang.org/protobuf/proto"
+)
+
+// sampleResponse builds a metadata event exercising the fields touched on the
+// hot subscribe path: nested entries, a repeated chunk list, a map, and byte
+// fields.
+func sampleResponse() *SubscribeMetadataResponse {
+	entry := func(name string) *Entry {
+		return &Entry{
+			Name:        name,
+			IsDirectory: false,
+			Chunks: []*FileChunk{
+				{
+					FileId:       "3,01637037d6",
+					Offset:       0,
+					Size:         1234,
+					ModifiedTsNs: 111,
+					ETag:         "abc",
+					Fid:          &FileId{VolumeId: 3, FileKey: 0x1637037, Cookie: 0xd6},
+					CipherKey:    []byte{1, 2, 3, 4},
+				},
+			},
+			Attributes: &FuseAttributes{
+				FileSize: 1234, Mtime: 222, FileMode: 0644, Uid: 1000, Gid: 1000,
+				GroupName: []string{"staff", "wheel"}, Md5: []byte{9, 8, 7},
+			},
+			Extended: map[string][]byte{
+				"foo": []byte("bar"),
+				"baz": []byte("qux"),
+				"k3":  []byte("v3"),
+			},
+			Content: []byte("hello"),
+		}
+	}
+	return &SubscribeMetadataResponse{
+		Directory: "/buckets/test/dir",
+		EventNotification: &EventNotification{
+			OldEntry:           entry("old.txt"),
+			NewEntry:           entry("new.txt"),
+			DeleteChunks:       true,
+			NewParentPath:      "/buckets/test/dir2",
+			IsFromOtherCluster: true,
+			Signatures:         []int32{101, 202, 303},
+		},
+		TsNs: 1234567890,
+	}
+}
+
+// eventWithChunks models a real create/update event: a file split into n chunks,
+// each a nested FileChunk+FileId message. n scales the number of nested messages
+// the decoder must allocate, which is where vtproto's edge over reflection grows.
+func eventWithChunks(n int) *SubscribeMetadataResponse {
+	chunks := make([]*FileChunk, n)
+	for i := 0; i < n; i++ {
+		chunks[i] = &FileChunk{
+			FileId:       fmt.Sprintf("%d,%08x%04x", i%64, i*7919, i%251),
+			Offset:       int64(i) * 8 << 20,
+			Size:         8 << 20,
+			ModifiedTsNs: 1700000000000000000 + int64(i),
+			ETag:         fmt.Sprintf("etag-%d", i),
+			Fid:          &FileId{VolumeId: uint32(i%64 + 1), FileKey: uint64(i * 7919), Cookie: uint32(i % 251)},
+		}
+	}
+	return &SubscribeMetadataResponse{
+		Directory: "/buckets/data/2026/07/15",
+		EventNotification: &EventNotification{
+			NewEntry: &Entry{
+				Name:       "object.bin",
+				Chunks:     chunks,
+				Attributes: &FuseAttributes{FileSize: uint64(n) * 8 << 20, Mtime: 1700000000, FileMode: 0644, Uid: 1000, Gid: 1000},
+			},
+			Signatures: []int32{12345},
+		},
+		TsNs: 1700000000000000000,
+	}
+}
+
+// TestVTWireCompatibility guards that the vtproto-generated MarshalVT/UnmarshalVT
+// stay wire-compatible with the reflection-based proto path in both directions,
+// so adopting VT on the metadata hot paths cannot silently corrupt events.
+func TestVTWireCompatibility(t *testing.T) {
+	orig := sampleResponse()
+
+	// vt marshal -> proto unmarshal
+	vtBytes, err := orig.MarshalVT()
+	if err != nil {
+		t.Fatalf("MarshalVT: %v", err)
+	}
+	var fromVT SubscribeMetadataResponse
+	if err := proto.Unmarshal(vtBytes, &fromVT); err != nil {
+		t.Fatalf("proto.Unmarshal(vtBytes): %v", err)
+	}
+	if !proto.Equal(orig, &fromVT) {
+		t.Fatalf("proto.Unmarshal(MarshalVT) diverged from original")
+	}
+
+	// proto marshal -> vt unmarshal
+	protoBytes, err := proto.Marshal(orig)
+	if err != nil {
+		t.Fatalf("proto.Marshal: %v", err)
+	}
+	var fromProto SubscribeMetadataResponse
+	if err := fromProto.UnmarshalVT(protoBytes); err != nil {
+		t.Fatalf("UnmarshalVT(protoBytes): %v", err)
+	}
+	if !proto.Equal(orig, &fromProto) {
+		t.Fatalf("UnmarshalVT(proto.Marshal) diverged from original")
+	}
+
+	// MarshalToSizedBufferVT into a pre-sized slice matches MarshalVT, the write
+	// path used by the log buffer.
+	sized := make([]byte, orig.SizeVT())
+	if _, err := orig.MarshalToSizedBufferVT(sized); err != nil {
+		t.Fatalf("MarshalToSizedBufferVT: %v", err)
+	}
+	var fromSized SubscribeMetadataResponse
+	if err := fromSized.UnmarshalVT(sized); err != nil {
+		t.Fatalf("UnmarshalVT(sized): %v", err)
+	}
+	if !proto.Equal(orig, &fromSized) {
+		t.Fatalf("MarshalToSizedBufferVT diverged from original")
+	}
+
+	// LogEntry carrying a marshaled event, mirroring the log buffer write path.
+	le := &LogEntry{TsNs: 7, PartitionKeyHash: 9, Data: vtBytes, Key: []byte("k"), Offset: 42}
+	leBytes, err := le.MarshalVT()
+	if err != nil {
+		t.Fatalf("LogEntry.MarshalVT: %v", err)
+	}
+	var leDecoded LogEntry
+	if err := proto.Unmarshal(leBytes, &leDecoded); err != nil {
+		t.Fatalf("proto.Unmarshal(LogEntry): %v", err)
+	}
+	if !proto.Equal(le, &leDecoded) {
+		t.Fatalf("LogEntry vt/proto roundtrip diverged")
+	}
+}
+
+// TestUTF8Validation documents the deliberate UTF-8 difference between the two
+// decoders. UnmarshalVT skips proto3's UTF-8 validation for string fields, so
+// the metadata delivery paths keep decoding SubscribeMetadataResponse with
+// proto.Unmarshal; only LogEntry (bytes-only) decodes use UnmarshalVT.
+func TestUTF8Validation(t *testing.T) {
+	// SubscribeMetadataResponse{Directory: "\xff"}: field 1 (string), wire type 2,
+	// length 1, byte 0xff.
+	malformed := []byte{0x0a, 0x01, 0xff}
+
+	var viaProto SubscribeMetadataResponse
+	if err := proto.Unmarshal(malformed, &viaProto); err == nil {
+		t.Fatal("proto.Unmarshal accepted invalid UTF-8 in a string field; expected rejection")
+	}
+
+	var viaVT SubscribeMetadataResponse
+	if err := viaVT.UnmarshalVT(malformed); err != nil {
+		t.Fatalf("UnmarshalVT rejected invalid UTF-8: %v", err)
+	}
+	if viaVT.Directory != "\xff" {
+		t.Fatalf("UnmarshalVT Directory = %q, want \\xff", viaVT.Directory)
+	}
+
+	// LogEntry carries only bytes fields, so the decoders agree on arbitrary bytes;
+	// that is why the log entry decode paths can use UnmarshalVT.
+	// LogEntry{Data: "\xff"}: field 3 (bytes), wire type 2, length 1, byte 0xff.
+	leBytes := []byte{0x1a, 0x01, 0xff}
+	var leProto LogEntry
+	if err := proto.Unmarshal(leBytes, &leProto); err != nil {
+		t.Fatalf("proto.Unmarshal LogEntry: %v", err)
+	}
+	var leVT LogEntry
+	if err := leVT.UnmarshalVT(leBytes); err != nil {
+		t.Fatalf("UnmarshalVT LogEntry: %v", err)
+	}
+	if !proto.Equal(&leProto, &leVT) {
+		t.Fatal("LogEntry proto/VT decode of a bytes field diverged")
+	}
+}
+
+var chunkCounts = []int{1, 8, 32, 128}
+
+// BenchmarkUnmarshal is the decode hot path: each subscriber decodes every
+// delivered event, so this cost is paid once per (event, subscriber).
+func BenchmarkUnmarshal(b *testing.B) {
+	for _, n := range chunkCounts {
+		data, _ := proto.Marshal(eventWithChunks(n))
+		b.Run(fmt.Sprintf("proto/chunks=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				var m SubscribeMetadataResponse
+				if err := proto.Unmarshal(data, &m); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+		b.Run(fmt.Sprintf("vt/chunks=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				var m SubscribeMetadataResponse
+				if err := m.UnmarshalVT(data); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkMarshal is the encode path: each metadata write marshals one LogEntry
+// payload.
+func BenchmarkMarshal(b *testing.B) {
+	for _, n := range chunkCounts {
+		ev := eventWithChunks(n)
+		b.Run(fmt.Sprintf("proto/chunks=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				if _, err := proto.Marshal(ev); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+		b.Run(fmt.Sprintf("vt/chunks=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				if _, err := ev.MarshalVT(); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkMarshalIntoBuffer mirrors the log buffer write path: marshal into a
+// reused destination buffer. proto.Marshal must allocate a fresh slice and copy;
+// SizeVT + MarshalToSizedBufferVT writes straight into the destination.
+func BenchmarkMarshalIntoBuffer(b *testing.B) {
+	for _, n := range chunkCounts {
+		ev := eventWithChunks(n)
+		dst := make([]byte, 0, ev.SizeVT()*2)
+		b.Run(fmt.Sprintf("proto+copy/chunks=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				data, err := proto.Marshal(ev)
+				if err != nil {
+					b.Fatal(err)
+				}
+				dst = dst[:len(data)]
+				copy(dst, data)
+			}
+		})
+		b.Run(fmt.Sprintf("vt-sized/chunks=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				size := ev.SizeVT()
+				dst = dst[:size]
+				if _, err := ev.MarshalToSizedBufferVT(dst); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}

weed/pb/filer_pb_direct_read.go

@@ -236,6 +236,8 @@ func readFilerFilesToChannel(
 
 func processOneLogEntry(logEntry *filer_pb.LogEntry, filter PathFilter, processEventFn ProcessMetadataFunc) (int64, error) {
 	event := &filer_pb.SubscribeMetadataResponse{}
+	// proto.Unmarshal (not UnmarshalVT) validates UTF-8 in string fields, so
+	// malformed metadata is skipped here instead of reaching path filtering.
 	if err := proto.Unmarshal(logEntry.Data, event); err != nil {
 		glog.Errorf("unmarshal log entry: %v", err)
 		return 0, nil // skip corrupt entries
@@ -329,7 +331,7 @@ func streamLogFileEntries(newReader LogFileReaderFn, chunks []*filer_pb.FileChun
 		}
 
 		logEntry := &filer_pb.LogEntry{}
-		if err := proto.Unmarshal(entryData, logEntry); err != nil {
+		if err := logEntry.UnmarshalVT(entryData); err != nil {
 			return err
 		}
 

weed/pb/filer_pb_direct_read_test.go

@@ -449,3 +449,29 @@ func TestReadLogFileRefsSingleFilerNotFoundSkips(t *testing.T) {
 		t.Fatalf("expected %d events after skipping one file, got %d", want, count)
 	}
 }
+
+// TestProcessOneLogEntrySkipsInvalidUTF8 guards that an event whose string field
+// is not valid UTF-8 is rejected by proto.Unmarshal and never reaches path
+// filtering or the consumer. UnmarshalVT would accept it, so the decode here
+// must stay on proto.Unmarshal.
+func TestProcessOneLogEntrySkipsInvalidUTF8(t *testing.T) {
+	// SubscribeMetadataResponse{Directory: "\xff"}: field 1 (string), wire type 2,
+	// length 1, byte 0xff — a syntactically valid message with an invalid-UTF-8
+	// string, which proto3 forbids.
+	logEntry := &filer_pb.LogEntry{Data: []byte{0x0a, 0x01, 0xff}}
+
+	called := false
+	tsNs, err := processOneLogEntry(logEntry, PathFilter{PathPrefix: "/"}, func(resp *filer_pb.SubscribeMetadataResponse) error {
+		called = true
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("processOneLogEntry err = %v, want nil (corrupt entries are skipped)", err)
+	}
+	if called {
+		t.Fatal("consumer was invoked for a malformed (invalid UTF-8) event; validation was bypassed")
+	}
+	if tsNs != 0 {
+		t.Fatalf("tsNs = %d, want 0 for a skipped entry", tsNs)
+	}
+}

weed/server/filer_grpc_server_sub_meta.go

@@ -515,6 +515,9 @@ func eachLogEntryFn(req *filer_pb.SubscribeMetadataRequest, sender metadataStrea
 			}
 		}
 		event := &filer_pb.SubscribeMetadataResponse{}
+		// proto.Unmarshal (not UnmarshalVT) validates UTF-8 in string fields, so
+		// malformed metadata is rejected here instead of reaching path filtering
+		// and subscribers.
 		if err := proto.Unmarshal(logEntry.Data, event); err != nil {
 			glog.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
 			return false, fmt.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %w", err)

weed/util/log_buffer/log_buffer.go

@@ -9,8 +9,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"google.golang.org/protobuf/proto"
-
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
@@ -302,13 +300,9 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) err
 		logBuffer.LastTsNs.Store(processingTsNs)
 	}
 
-	logEntryData, err := proto.Marshal(logEntry)
-	if err != nil {
-		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
-		glog.Errorf("%v", marshalErr)
-		return marshalErr
-	}
-	size := len(logEntryData)
+	// Size is computed without allocating; the entry is marshaled straight into
+	// the buffer below via MarshalToSizedBufferVT.
+	size := logEntry.SizeVT()
 
 	if logBuffer.pos == 0 {
 		logBuffer.startTime = ts
@@ -352,10 +346,17 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) err
 	}
 	logBuffer.stopTime = ts
 
+	// Marshal directly into the buffer, avoiding an intermediate slice and copy.
+	// On the (practically impossible) error the entry is dropped before idx/pos
+	// are advanced, leaving the buffer consistent.
+	if _, err := logEntry.MarshalToSizedBufferVT(logBuffer.buf[logBuffer.pos+4 : logBuffer.pos+4+size]); err != nil {
+		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
+		glog.Errorf("%v", marshalErr)
+		return marshalErr
+	}
 	logBuffer.idx = append(logBuffer.idx, logBuffer.pos)
 	util.Uint32toBytes(logBuffer.sizeBuf, uint32(size))
 	copy(logBuffer.buf[logBuffer.pos:logBuffer.pos+4], logBuffer.sizeBuf)
-	copy(logBuffer.buf[logBuffer.pos+4:logBuffer.pos+4+size], logEntryData)
 	logBuffer.pos += size + 4
 
 	logBuffer.offset++
@@ -416,15 +417,9 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	// Note: This also enables AddToBuffer to work correctly with Kafka-style offset-based reads
 	logEntry.Offset = logBuffer.offset
 
-	// Marshal with correct timestamp and offset
-	logEntryData, err := proto.Marshal(logEntry)
-	if err != nil {
-		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
-		glog.Errorf("%v", marshalErr)
-		return marshalErr
-	}
-
-	size := len(logEntryData)
+	// Size is computed without allocating; the entry is marshaled straight into
+	// the buffer below via MarshalToSizedBufferVT.
+	size := logEntry.SizeVT()
 
 	if logBuffer.pos == 0 {
 		logBuffer.startTime = ts
@@ -466,10 +461,17 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	}
 	logBuffer.stopTime = ts
 
+	// Marshal directly into the buffer, avoiding an intermediate slice and copy.
+	// On the (practically impossible) error the entry is dropped before idx/pos
+	// are advanced, leaving the buffer consistent.
+	if _, err := logEntry.MarshalToSizedBufferVT(logBuffer.buf[logBuffer.pos+4 : logBuffer.pos+4+size]); err != nil {
+		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
+		glog.Errorf("%v", marshalErr)
+		return marshalErr
+	}
 	logBuffer.idx = append(logBuffer.idx, logBuffer.pos)
 	util.Uint32toBytes(logBuffer.sizeBuf, uint32(size))
 	copy(logBuffer.buf[logBuffer.pos:logBuffer.pos+4], logBuffer.sizeBuf)
-	copy(logBuffer.buf[logBuffer.pos+4:logBuffer.pos+4+size], logEntryData)
 	logBuffer.pos += size + 4
 
 	logBuffer.offset++

weed/util/log_buffer/log_read_stateless.go

@@ -422,7 +422,7 @@ func parseMessagesFromBuffer(buf []byte, startOffset int64, maxMessages int, max
 		// Parse message
 		entryData := buf[pos+4 : pos+4+int(size)]
 		logEntry := &filer_pb.LogEntry{}
-		if err = proto.Unmarshal(entryData, logEntry); err != nil {
+		if err = logEntry.UnmarshalVT(entryData); err != nil {
 			glog.Warningf("[parseMessages] Failed to unmarshal message: %v", err)
 			pos += 4 + int(size)
 			continue