refactor: replace proto serialization with JSON and migrate callsites

config/config.go

@@ -70,6 +70,9 @@ func Parse(configFilePath string) (*Config, error) {
 	if config.Service.WorkerPoolSize <= 0 {
 		return nil, fmt.Errorf("service.worker_pool_size must be > 0, got %d", config.Service.WorkerPoolSize)
 	}
+	if config.Service.MaxMessageBytes <= 0 {
+		config.Service.MaxMessageBytes = DefaultMaxMessageBytes
+	}
 	config.repositoryByRemote = make(map[string]*RepositoryConfig, len(config.Repository))
 	for i := range config.Repository {
 		remote := config.Repository[i].Remote

config/service_config.go

@@ -16,23 +16,12 @@ package config
 
 // ServiceConfig holds operational configuration for the Tango service.
 type ServiceConfig struct {
-	WorkerPoolSize       int         `yaml:"worker_pool_size"`        // number of worker workspaces per repo
-	RepoManagerClonePath string      `yaml:"repo_manager_clone_path"` // root directory for origin repo clones
-	WorkerRootPath       string      `yaml:"worker_root_path"`        // root directory for worker workspace checkouts; defaults to repo_manager_clone_path/.workers
-	Chunking             ChunkConfig `yaml:"chunking"`                // streaming chunk sizes; zero values fall back to package defaults
+	WorkerPoolSize       int    `yaml:"worker_pool_size"`        // number of worker workspaces per repo
+	RepoManagerClonePath string `yaml:"repo_manager_clone_path"` // root directory for origin repo clones
+	WorkerRootPath       string `yaml:"worker_root_path"`        // root directory for worker workspace checkouts; defaults to repo_manager_clone_path/.workers
+	MaxMessageBytes      int    `yaml:"max_message_bytes"`       // max serialized bytes per streamed gRPC message; 0 → DefaultMaxMessageBytes
 }
 
-// ChunkConfig controls the number of entries per gRPC stream message.
-// All fields are optional; a zero value means "use the package default".
-// Tune these when a monorepo's per-target size causes messages to approach
-// the 64MB default gRPC per-message limit.
-type ChunkConfig struct {
-	// TargetChunkSize is the max number of OptimizedTarget entries per stream message.
-	TargetChunkSize int `yaml:"target_chunk_size"`
-	// ChangedTargetChunkSize is the max number of ChangedTarget entries per stream message.
-	// ChangedTarget carries both old and new targets (~2× the size of a regular target).
-	ChangedTargetChunkSize int `yaml:"changed_target_chunk_size"`
-	// MetadataMapChunkSize is the max number of entries per metadata map chunk.
-	// Applies to target_id_mapping and attribute_string_value_mapping.
-	MetadataMapChunkSize int `yaml:"metadata_map_chunk_size"`
-}
+// DefaultMaxMessageBytes is the fallback max serialized size per streamed
+// message (~4.25 MB), well under the 64 MB default gRPC limit.
+const DefaultMaxMessageBytes = 4_250_000

controller/BUILD.bazel

@@ -17,12 +17,12 @@ go_library(
     deps = [
         "//config",
         "//core/cachekey",
-        "//core/common",
         "//core/errors",
         "//core/storage",
         "//entity",
         "//internal/mapper",
         "//internal/mapper/idmapper",
+        "//internal/streaming",
         "//internal/targetdiff",
         "//internal/url",
         "//observability/metrics",
@@ -47,15 +47,14 @@ go_test(
     ],
     embed = [":controller"],
     deps = [
-        "//core/common",
+        "//config",
         "//core/storage",
         "//core/storage/storagemock",
         "//entity",
         "//observability/metrics",
         "//orchestrator/orchestratormock",
         "//tangopb",
         "//tangopb/tangopbmock",
-        "@com_github_gogo_protobuf//io",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_uber_go_tally//:tally",

controller/controller.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/uber-go/tally"
 	"github.com/uber/tango/config"
-	"github.com/uber/tango/core/common"
 	"github.com/uber/tango/core/storage"
 	"github.com/uber/tango/observability/metrics"
 	"github.com/uber/tango/orchestrator"
@@ -31,21 +30,19 @@ import (
 // Params are the parameters for the controller.
 type Params struct {
 	fx.In
-	Logger       *zap.Logger
-	Storage      storage.Storage
-	Orchestrator orchestrator.Orchestrator
-	Scope        tally.Scope        `optional:"true"`
-	ChunkConfig  config.ChunkConfig `optional:"true"`
+	Logger          *zap.Logger
+	Storage         storage.Storage
+	Orchestrator    orchestrator.Orchestrator
+	Scope           tally.Scope `optional:"true"`
+	MaxMessageBytes int         `optional:"true"`
 }
 
 type controller struct {
-	logger                 *zap.Logger
-	storage                storage.Storage
-	orchestrator           orchestrator.Orchestrator
-	emitter                *metrics.Emitter
-	targetChunkSize        int
-	changedTargetChunkSize int
-	metadataMapChunkSize   int
+	logger          *zap.Logger
+	storage         storage.Storage
+	orchestrator    orchestrator.Orchestrator
+	emitter         *metrics.Emitter
+	maxMessageBytes int
 
 	// appCtx is the application lifetime; cancel it on process shutdown.
 	// Used by linkRequestCtx and any fire-and-forget goroutines so they
@@ -57,27 +54,17 @@ type controller struct {
 // shutdown to abort background work.
 func NewController(appCtx context.Context, p Params) pb.TangoYARPCServer {
 	emitter := metrics.New(p.Scope).SubScope("controller")
-	targetChunkSize := p.ChunkConfig.TargetChunkSize
-	if targetChunkSize <= 0 {
-		targetChunkSize = common.DefaultTargetChunkSize
-	}
-	changedTargetChunkSize := p.ChunkConfig.ChangedTargetChunkSize
-	if changedTargetChunkSize <= 0 {
-		changedTargetChunkSize = common.DefaultChangedTargetChunkSize
-	}
-	metadataMapChunkSize := p.ChunkConfig.MetadataMapChunkSize
-	if metadataMapChunkSize <= 0 {
-		metadataMapChunkSize = common.DefaultMetadataMapChunkSize
+	maxMessageBytes := p.MaxMessageBytes
+	if maxMessageBytes <= 0 {
+		maxMessageBytes = config.DefaultMaxMessageBytes
 	}
 	return &controller{
-		logger:                 p.Logger,
-		storage:                p.Storage,
-		orchestrator:           p.Orchestrator,
-		emitter:                emitter,
-		targetChunkSize:        targetChunkSize,
-		changedTargetChunkSize: changedTargetChunkSize,
-		metadataMapChunkSize:   metadataMapChunkSize,
-		appCtx:                 appCtx,
+		logger:          p.Logger,
+		storage:         p.Storage,
+		orchestrator:    p.Orchestrator,
+		emitter:         emitter,
+		maxMessageBytes: maxMessageBytes,
+		appCtx:          appCtx,
 	}
 }