fix(config): validate storage and unknown fields in Parse, wire max_message_bytes (audit #12)

config/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "config",
@@ -12,3 +12,13 @@ go_library(
     visibility = ["//visibility:public"],
     deps = ["@com_github_goccy_go_yaml//:go-yaml"],
 )
+
+go_test(
+    name = "config_test",
+    srcs = ["config_test.go"],
+    embed = [":config"],
+    deps = [
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
+)

config/config.go

@@ -15,6 +15,7 @@
 package config
 
 import (
+	"bytes"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -50,17 +51,35 @@ func Parse(configFilePath string) (*Config, error) {
 	if err != nil {
 		return nil, err
 	}
+	return ParseBytes(yamlBytes)
+}
+
+// ParseBytes parses the full configuration from raw YAML bytes.
+// Unknown YAML fields are rejected.
+func ParseBytes(yamlBytes []byte) (*Config, error) {
 	var config Config
-	if err := yaml.Unmarshal(yamlBytes, &config); err != nil {
+	dec := yaml.NewDecoder(bytes.NewReader(yamlBytes), yaml.DisallowUnknownField())
+	if err := dec.Decode(&config); err != nil {
 		return nil, err
 	}
-	// Default to memory storage if not specified
-	if config.Storage.Type == "" {
+
+	// --- storage validation ---
+	switch config.Storage.Type {
+	case StorageTypeMemory, "":
 		config.Storage.Type = StorageTypeMemory
+	case StorageTypeDisk:
+		if config.Storage.Disk == nil || config.Storage.Disk.RootPath == "" {
+			return nil, fmt.Errorf("storage.disk.root_path must be set when storage type is %q", StorageTypeDisk)
+		}
+	default:
+		return nil, fmt.Errorf("unsupported storage type: %q (supported: %q, %q)", config.Storage.Type, StorageTypeMemory, StorageTypeDisk)
 	}
+
+	// --- service validation and defaults ---
 	if config.Service.WorkerRootPath != "" && config.Service.RepoManagerClonePath == "" {
 		return nil, fmt.Errorf("service.repo_manager_clone_path must be set when worker_root_path is specified")
 	}
+	// Default: os.TempDir()/tango-repo-manager.
 	if config.Service.RepoManagerClonePath == "" {
 		config.Service.RepoManagerClonePath = filepath.Join(os.TempDir(), "tango-repo-manager")
 	}
@@ -73,6 +92,8 @@ func Parse(configFilePath string) (*Config, error) {
 	if config.Service.MaxMessageBytes <= 0 {
 		config.Service.MaxMessageBytes = DefaultMaxMessageBytes
 	}
+
+	// --- repository validation and defaults ---
 	config.repositoryByRemote = make(map[string]*RepositoryConfig, len(config.Repository))
 	for i := range config.Repository {
 		remote := config.Repository[i].Remote
@@ -82,6 +103,11 @@ func Parse(configFilePath string) (*Config, error) {
 		if _, exists := config.repositoryByRemote[remote]; exists {
 			return nil, fmt.Errorf("duplicate repository remote %q", remote)
 		}
+		// Default query_timeout: 900 seconds (15 minutes), matching the
+		// core/bazel package's _queryTimeout constant.
+		if config.Repository[i].QueryTimeout <= 0 {
+			config.Repository[i].QueryTimeout = DefaultQueryTimeoutSeconds
+		}
 		config.repositoryByRemote[remote] = &config.Repository[i]
 	}
 	return &config, nil

config/config_test.go

@@ -0,0 +1,243 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// minimal returns a valid minimal YAML config string.
+func minimal() string {
+	return `
+repository:
+  - remote: "https://example.com/repo.git"
+service:
+  worker_pool_size: 2
+`
+}
+
+func TestParseBytes_Defaults(t *testing.T) {
+	cfg, err := ParseBytes([]byte(minimal()))
+	require.NoError(t, err)
+
+	assert.Equal(t, StorageTypeMemory, cfg.Storage.Type, "storage type should default to memory")
+	assert.Equal(t, DefaultMaxMessageBytes, cfg.Service.MaxMessageBytes, "max_message_bytes should default")
+	assert.Contains(t, cfg.Service.RepoManagerClonePath, "tango-repo-manager", "clone path should default to temp dir")
+	assert.Contains(t, cfg.Service.WorkerRootPath, ".workers", "worker root should default under clone path")
+	assert.Equal(t, DefaultQueryTimeoutSeconds, cfg.Repository[0].QueryTimeout, "query_timeout should default to 900")
+}
+
+func TestParseBytes_ExplicitValues(t *testing.T) {
+	yamlStr := `
+repository:
+  - remote: "https://example.com/repo.git"
+    query_timeout: 60
+storage:
+  type: "memory"
+service:
+  worker_pool_size: 4
+  repo_manager_clone_path: "/custom/clone"
+  worker_root_path: "/custom/workers"
+  max_message_bytes: 1000000
+`
+	cfg, err := ParseBytes([]byte(yamlStr))
+	require.NoError(t, err)
+
+	assert.Equal(t, StorageTypeMemory, cfg.Storage.Type)
+	assert.Equal(t, int64(60), cfg.Repository[0].QueryTimeout)
+	assert.Equal(t, "/custom/clone", cfg.Service.RepoManagerClonePath)
+	assert.Equal(t, "/custom/workers", cfg.Service.WorkerRootPath)
+	assert.Equal(t, 1000000, cfg.Service.MaxMessageBytes)
+}
+
+func TestParseBytes_StorageValidation(t *testing.T) {
+	tests := []struct {
+		name    string
+		yaml    string
+		wantErr bool
+	}{
+		{
+			name: "memory explicit",
+			yaml: `
+storage:
+  type: "memory"
+repository:
+  - remote: "https://example.com/r.git"
+service:
+  worker_pool_size: 1
+`,
+		},
+		{
+			name: "empty defaults to memory",
+			yaml: `
+repository:
+  - remote: "https://example.com/r.git"
+service:
+  worker_pool_size: 1
+`,
+		},
+		{
+			name: "disk with root_path",
+			yaml: `
+storage:
+  type: "disk"
+  disk:
+    root_path: "/tmp/store"
+repository:
+  - remote: "https://example.com/r.git"
+service:
+  worker_pool_size: 1
+`,
+		},
+		{
+			name:    "disk without root_path",
+			wantErr: true,
+			yaml: `
+storage:
+  type: "disk"
+repository:
+  - remote: "https://example.com/r.git"
+service:
+  worker_pool_size: 1
+`,
+		},
+		{
+			name:    "disk with empty root_path",
+			wantErr: true,
+			yaml: `
+storage:
+  type: "disk"
+  disk:
+    root_path: ""
+repository:
+  - remote: "https://example.com/r.git"
+service:
+  worker_pool_size: 1
+`,
+		},
+		{
+			name:    "unknown storage type",
+			wantErr: true,
+			yaml: `
+storage:
+  type: "s3"
+repository:
+  - remote: "https://example.com/r.git"
+service:
+  worker_pool_size: 1
+`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := ParseBytes([]byte(tt.yaml))
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestParseBytes_UnknownFieldsRejected(t *testing.T) {
+	yamlStr := `
+repository:
+  - remote: "https://example.com/repo.git"
+service:
+  worker_pool_size: 1
+  totally_bogus_field: true
+`
+	_, err := ParseBytes([]byte(yamlStr))
+	require.Error(t, err, "unknown YAML fields should be rejected")
+}
+
+func TestParseBytes_WorkerPoolSizeRequired(t *testing.T) {
+	yamlStr := `
+repository:
+  - remote: "https://example.com/repo.git"
+service:
+  worker_pool_size: 0
+`
+	_, err := ParseBytes([]byte(yamlStr))
+	require.Error(t, err)
+}
+
+func TestParseBytes_EmptyRemoteRejected(t *testing.T) {
+	yamlStr := `
+repository:
+  - remote: ""
+service:
+  worker_pool_size: 1
+`
+	_, err := ParseBytes([]byte(yamlStr))
+	require.Error(t, err)
+}
+
+func TestParseBytes_DuplicateRemoteRejected(t *testing.T) {
+	yamlStr := `
+repository:
+  - remote: "https://example.com/repo.git"
+  - remote: "https://example.com/repo.git"
+service:
+  worker_pool_size: 1
+`
+	_, err := ParseBytes([]byte(yamlStr))
+	require.Error(t, err)
+}
+
+func TestParseBytes_WorkerRootPathRequiresClonePath(t *testing.T) {
+	yamlStr := `
+repository:
+  - remote: "https://example.com/repo.git"
+service:
+  worker_pool_size: 1
+  worker_root_path: "/some/path"
+`
+	_, err := ParseBytes([]byte(yamlStr))
+	require.Error(t, err)
+}
+
+func TestParse_FileNotFound(t *testing.T) {
+	_, err := Parse("/nonexistent/path/config.yaml")
+	require.Error(t, err)
+}
+
+func TestParse_FromFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "config.yaml")
+	require.NoError(t, os.WriteFile(path, []byte(minimal()), 0o644))
+
+	cfg, err := Parse(path)
+	require.NoError(t, err)
+	assert.Equal(t, StorageTypeMemory, cfg.Storage.Type)
+}
+
+func TestGetRepositoryConfig(t *testing.T) {
+	cfg, err := ParseBytes([]byte(minimal()))
+	require.NoError(t, err)
+
+	repo, ok := cfg.GetRepositoryConfig("https://example.com/repo.git")
+	assert.True(t, ok)
+	assert.Equal(t, "https://example.com/repo.git", repo.Remote)
+
+	_, ok = cfg.GetRepositoryConfig("https://missing.com/repo.git")
+	assert.False(t, ok)
+}

config/repository_config.go

@@ -14,6 +14,10 @@
 
 package config
 
+// DefaultQueryTimeoutSeconds is the default query_timeout applied when the
+// field is unset or zero (15 minutes), matching core/bazel's _queryTimeout.
+const DefaultQueryTimeoutSeconds int64 = 900
+
 // RepositoryConfig holds configuration for a single repository.
 type RepositoryConfig struct {
 	Remote                 string   `yaml:"remote"`
@@ -22,8 +26,9 @@ type RepositoryConfig struct {
 	ExcludeExternalTargets bool     `yaml:"exclude_external_targets"`
 	BzlmodEnabled          bool     `yaml:"bzlmod_enabled"`
 	BazelCommand           string   `yaml:"bazel_command"`
-	QueryTimeout           int64    `yaml:"query_timeout"` // in seconds
-	BazelExtraArgs         []string `yaml:"bazel_extra_args"`
+	// QueryTimeout is the Bazel query timeout in seconds. Defaults to DefaultQueryTimeoutSeconds (900, i.e. 15 minutes).
+	QueryTimeout   int64    `yaml:"query_timeout"`
+	BazelExtraArgs []string `yaml:"bazel_extra_args"`
 	// BazelStartupOptions are Bazel startup flags placed before the `query` subcommand (e.g. "--batch"); empty by default.
 	BazelStartupOptions []string `yaml:"bazel_startup_options"`
 	StreamBazelLogs     bool     `yaml:"stream_bazel_logs"`

config/service_config.go

@@ -16,10 +16,15 @@ 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
-	MaxMessageBytes      int    `yaml:"max_message_bytes"`       // max serialized bytes per streamed gRPC message; 0 → DefaultMaxMessageBytes
+	WorkerPoolSize int `yaml:"worker_pool_size"` // number of worker workspaces per repo
+	// RepoManagerClonePath is the root directory for origin repo clones.
+	// Defaults to os.TempDir()/tango-repo-manager.
+	// Must be set explicitly when worker_root_path is specified.
+	RepoManagerClonePath string `yaml:"repo_manager_clone_path"`
+	// WorkerRootPath is the root directory for worker workspace checkouts.
+	// Defaults to repo_manager_clone_path/.workers.
+	WorkerRootPath  string `yaml:"worker_root_path"`
+	MaxMessageBytes int    `yaml:"max_message_bytes"` // max serialized bytes per streamed gRPC message; 0 → DefaultMaxMessageBytes
 }
 
 // DefaultMaxMessageBytes is the fallback max serialized size per streamed