upstream-seaweedfs/weed/pb/filer_pb/filer_vtproto_test.go

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)
				}
			}
		})
	}
}

// The log position fields must survive VT round-trips, or a VT-marshaled
// response silently loses the version the mount fences handles with.
func TestVtProtoPreservesLogTsNs(t *testing.T) {
	messages := []interface {
		MarshalVT() ([]byte, error)
	}{
		&LookupDirectoryEntryResponse{LogTsNs: 42},
		&CacheRemoteObjectToLocalClusterResponse{LogTsNs: 43},
		&CreateEntryResponse{LogTsNs: 44},
		&UpdateEntryResponse{LogTsNs: 45},
	}
	for i, m := range messages {
		data, err := m.MarshalVT()
		if err != nil {
			t.Fatalf("MarshalVT %d: %v", i, err)
		}
		switch v := m.(type) {
		case *LookupDirectoryEntryResponse:
			out := &LookupDirectoryEntryResponse{}
			if err := out.UnmarshalVT(data); err != nil || out.LogTsNs != v.LogTsNs {
				t.Fatalf("LookupDirectoryEntryResponse round-trip LogTsNs = %d, %v; want %d", out.LogTsNs, err, v.LogTsNs)
			}
		case *CacheRemoteObjectToLocalClusterResponse:
			out := &CacheRemoteObjectToLocalClusterResponse{}
			if err := out.UnmarshalVT(data); err != nil || out.LogTsNs != v.LogTsNs {
				t.Fatalf("CacheRemoteObjectToLocalClusterResponse round-trip LogTsNs = %d, %v; want %d", out.LogTsNs, err, v.LogTsNs)
			}
		case *CreateEntryResponse:
			out := &CreateEntryResponse{}
			if err := out.UnmarshalVT(data); err != nil || out.LogTsNs != v.LogTsNs {
				t.Fatalf("CreateEntryResponse round-trip LogTsNs = %d, %v; want %d", out.LogTsNs, err, v.LogTsNs)
			}
		case *UpdateEntryResponse:
			out := &UpdateEntryResponse{}
			if err := out.UnmarshalVT(data); err != nil || out.LogTsNs != v.LogTsNs {
				t.Fatalf("UpdateEntryResponse round-trip LogTsNs = %d, %v; want %d", out.LogTsNs, err, v.LogTsNs)
			}
		}
	}
}