refactor: migrate callsites to entity types and consolidate config

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

@@ -14,13 +14,13 @@ go_library(
     importpath = "github.com/uber/tango/controller",
     visibility = ["//visibility:public"],
     deps = [
-        "//config",
         "//core/common",
         "//core/storage",
         "//entity",
         "//internal/cachekey",
         "//internal/mapper",
         "//internal/mapper/idmapper",
+        "//internal/streaming",
         "//internal/targetdiff",
         "//orchestrator",
         "//tangopb",
@@ -49,7 +49,6 @@ go_test(
         "//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,8 +19,6 @@ import (
 	"time"
 
 	"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/orchestrator"
 	pb "github.com/uber/tango/tangopb"
@@ -31,25 +29,25 @@ 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"`
 }
 
 // _totalDurationBuckets covers 0–15 minutes in 10-second linear intervals.
 var _totalDurationBuckets = tally.MustMakeLinearDurationBuckets(10*time.Second, 10*time.Second, 90)
 
+const _defaultMaxMessageBytes = 4_250_000
+
 type controller struct {
-	logger                 *zap.Logger
-	storage                storage.Storage
-	orchestrator           orchestrator.Orchestrator
-	scope                  tally.Scope
-	targetChunkSize        int
-	changedTargetChunkSize int
-	metadataMapChunkSize   int
-	totalDurationBuckets   tally.Buckets
+	logger               *zap.Logger
+	storage              storage.Storage
+	orchestrator         orchestrator.Orchestrator
+	scope                tally.Scope
+	maxMessageBytes      int
+	totalDurationBuckets tally.Buckets
 
 	// appCtx is the application lifetime; cancel it on process shutdown.
 	// Used by linkRequestCtx and any fire-and-forget goroutines so they
@@ -64,28 +62,18 @@ func NewController(appCtx context.Context, p Params) pb.TangoYARPCServer {
 	if scope == nil {
 		scope = tally.NoopScope
 	}
-	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 = _defaultMaxMessageBytes
 	}
 	return &controller{
-		logger:                 p.Logger,
-		storage:                p.Storage,
-		orchestrator:           p.Orchestrator,
-		scope:                  scope.SubScope("controller"),
-		targetChunkSize:        targetChunkSize,
-		changedTargetChunkSize: changedTargetChunkSize,
-		metadataMapChunkSize:   metadataMapChunkSize,
-		totalDurationBuckets:   _totalDurationBuckets,
-		appCtx:                 appCtx,
+		logger:               p.Logger,
+		storage:              p.Storage,
+		orchestrator:         p.Orchestrator,
+		scope:                scope.SubScope("controller"),
+		maxMessageBytes:      maxMessageBytes,
+		totalDurationBuckets: _totalDurationBuckets,
+		appCtx:               appCtx,
 	}
 }
 
@@ -98,12 +86,7 @@ func NewController(appCtx context.Context, p Params) pb.TangoYARPCServer {
 // The returned cancel function MUST be deferred; it releases the
 // context.AfterFunc handle so we do not leak a watcher past the request.
 func (c *controller) linkRequestCtx(reqCtx context.Context) (context.Context, context.CancelFunc) {
-	// Derive a per-request ctx whose cancel only affects this ctx and its
-	// children — it never propagates up to reqCtx.
 	ctx, cancel := context.WithCancel(reqCtx)
-	// Register a one-shot watcher that cancels the derived ctx if appCtx fires.
-	// AfterFunc only observes appCtx; it never cancels it. stop() deregisters
-	// the watcher so the closure is not retained past the request.
 	stop := context.AfterFunc(c.appCtx, cancel)
 	return ctx, func() {
 		stop()