fix: validate serve limits before opening store

Author: sunrioa

cmd/rin/config_validation.go

@@ -8,6 +8,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/sunrioa/rin/generation"
+	"github.com/sunrioa/rin/jobs"
 	rinruntime "github.com/sunrioa/rin/runtime"
 )
 
@@ -163,6 +165,8 @@ type serveConfiguration struct {
 	scrubInterval             time.Duration
 	scrubTimeout              time.Duration
 	scrubMaxEvents            int
+	jobConfig                 jobs.Config
+	generationConfig          generation.Config
 }
 
 func validateServeConfiguration(config serveConfiguration) error {
@@ -198,7 +202,22 @@ func validateServeConfiguration(config serveConfiguration) error {
 			"scrub-max-events must be between 1 and %d",
 			rinruntime.MaxScrubEventBudget,
 		)
-	default:
-		return nil
 	}
+	if err := rinruntime.ValidateEngineOptions(rinruntime.EngineOptions{
+		SessionSoftLimitBytes:  config.sessionSoftLimitBytes,
+		SessionHardLimitBytes:  config.sessionHardLimitBytes,
+		MaxSessionStateBytes:   config.maxSessionStateBytes,
+		MaxTransferBytes:       config.maxTransferBytes,
+		MaxTransferEvents:      config.maxTransferEvents,
+		MaxConcurrentTransfers: config.maxConcurrentTransfers,
+	}); err != nil {
+		return fmt.Errorf("invalid Runtime limits: %w", err)
+	}
+	if err := jobs.ValidateConfig(config.jobConfig); err != nil {
+		return fmt.Errorf("invalid Proposal Job limits: %w", err)
+	}
+	if err := generation.ValidateConfig(config.generationConfig); err != nil {
+		return fmt.Errorf("invalid Generation limits: %w", err)
+	}
+	return nil
 }

cmd/rin/main.go

@@ -160,6 +160,32 @@ Serve options:
 	if err := validateServeEnvironment(); err != nil {
 		return err
 	}
+	jobConfig := jobs.Config{
+		Workers:   envInt("RIN_JOB_WORKERS", 2),
+		QueueSize: envInt("RIN_JOB_QUEUE_SIZE", 64),
+		MaxJobs:   envInt("RIN_JOB_MAX_RETAINED", 512),
+		JobTTL:    envDuration("RIN_JOB_TTL", 30*time.Minute),
+	}
+	generationConfig := generation.Config{
+		Workers:      envInt("RIN_GENERATION_WORKERS", 2),
+		QueueSize:    envInt("RIN_GENERATION_QUEUE_SIZE", 64),
+		MaxJobs:      envInt("RIN_GENERATION_MAX_RETAINED", 512),
+		JobTTL:       envDuration("RIN_GENERATION_JOB_TTL", 30*time.Minute),
+		CacheEntries: envInt("RIN_GENERATION_CACHE_ENTRIES", 256),
+		CacheTTL:     envDuration("RIN_GENERATION_CACHE_TTL", 30*time.Minute),
+		MaxOutputBytes: envInt(
+			"RIN_GENERATION_MAX_OUTPUT_BYTES",
+			512*1024,
+		),
+		MaxRetainedBytes: envUint64(
+			"RIN_GENERATION_MAX_RETAINED_BYTES",
+			64<<20,
+		),
+		CleanupInterval: envDuration(
+			"RIN_GENERATION_CLEANUP_INTERVAL",
+			time.Minute,
+		),
+	}
 	if err := validateServeConfiguration(serveConfiguration{
 		maxBodyBytes:              *maxBody,
 		sessionSoftLimitBytes:     *sessionSoftLimit,
@@ -174,6 +200,8 @@ Serve options:
 		scrubInterval:             *scrubInterval,
 		scrubTimeout:              *scrubTimeout,
 		scrubMaxEvents:            *scrubMaxEvents,
+		jobConfig:                 jobConfig,
+		generationConfig:          generationConfig,
 	}); err != nil {
 		return err
 	}
@@ -217,29 +245,16 @@ Serve options:
 		defer cancel()
 		resultErr = errors.Join(resultErr, engine.Close(closeContext))
 	}()
-	jobManager, err := jobs.New(engine, jobs.Config{
-		Workers: envInt("RIN_JOB_WORKERS", 2), QueueSize: envInt("RIN_JOB_QUEUE_SIZE", 64),
-		MaxJobs: envInt("RIN_JOB_MAX_RETAINED", 512), JobTTL: envDuration("RIN_JOB_TTL", 30*time.Minute),
-	})
+	jobManager, err := jobs.New(engine, jobConfig)
 	if err != nil {
 		return err
 	}
 	var generationManager *generation.Manager
 	if modelRuntime.GenerationProvider != nil {
-		generationManager, err = generation.New(modelRuntime.GenerationProvider, generation.Config{
-			Workers: envInt("RIN_GENERATION_WORKERS", 2), QueueSize: envInt("RIN_GENERATION_QUEUE_SIZE", 64),
-			MaxJobs: envInt("RIN_GENERATION_MAX_RETAINED", 512), JobTTL: envDuration("RIN_GENERATION_JOB_TTL", 30*time.Minute),
-			CacheEntries: envInt("RIN_GENERATION_CACHE_ENTRIES", 256), CacheTTL: envDuration("RIN_GENERATION_CACHE_TTL", 30*time.Minute),
-			MaxOutputBytes: envInt("RIN_GENERATION_MAX_OUTPUT_BYTES", 512*1024),
-			MaxRetainedBytes: envUint64(
-				"RIN_GENERATION_MAX_RETAINED_BYTES",
-				64<<20,
-			),
-			CleanupInterval: envDuration(
-				"RIN_GENERATION_CLEANUP_INTERVAL",
-				time.Minute,
-			),
-		})
+		generationManager, err = generation.New(
+			modelRuntime.GenerationProvider,
+			generationConfig,
+		)
 		if err != nil {
 			closeContext, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 			defer cancel()

cmd/rin/main_test.go

@@ -348,6 +348,44 @@ func TestValidateServeConfigurationRejectsExplicitFallbackValues(
 				config.scrubMaxEvents = rinruntime.MaxScrubEventBudget + 1
 			},
 		},
+		{
+			name: "Session State upper bound",
+			mutate: func(config *serveConfiguration) {
+				config.maxSessionStateBytes =
+					rinruntime.MaxConfigurableSessionStateBytes + 1
+			},
+		},
+		{
+			name: "Transfer byte upper bound",
+			mutate: func(config *serveConfiguration) {
+				config.maxTransferBytes = (1 << 40) + 1
+			},
+		},
+		{
+			name: "Transfer event JSON ceiling",
+			mutate: func(config *serveConfiguration) {
+				config.maxTransferEvents =
+					uint64(protocol.MaxJSONSafeInteger) + 1
+			},
+		},
+		{
+			name: "Transfer concurrency upper bound",
+			mutate: func(config *serveConfiguration) {
+				config.maxConcurrentTransfers = 65
+			},
+		},
+		{
+			name: "Proposal Job workers",
+			mutate: func(config *serveConfiguration) {
+				config.jobConfig.Workers = 33
+			},
+		},
+		{
+			name: "Generation output",
+			mutate: func(config *serveConfiguration) {
+				config.generationConfig.MaxOutputBytes = 4*1024*1024 + 1
+			},
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
@@ -363,6 +401,89 @@ func TestValidateServeConfigurationRejectsExplicitFallbackValues(
 	}
 }
 
+func TestInvalidServeLimitsDoNotTouchDataDirectory(t *testing.T) {
+	tests := []struct {
+		name string
+		env  string
+		args []string
+	}{
+		{
+			name: "Session State",
+			args: []string{
+				"-session-state-max-bytes",
+				fmt.Sprint(rinruntime.MaxConfigurableSessionStateBytes + 1),
+			},
+		},
+		{
+			name: "Transfer bytes",
+			args: []string{"-transfer-max-bytes", fmt.Sprint((1 << 40) + 1)},
+		},
+		{
+			name: "Transfer concurrency",
+			args: []string{"-transfer-max-concurrent", "65"},
+		},
+		{
+			name: "Proposal Job workers",
+			env:  "RIN_JOB_WORKERS=33",
+		},
+		{
+			name: "Generation output",
+			env:  "RIN_GENERATION_MAX_OUTPUT_BYTES=4194305",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			clearServeEnvironment(t)
+			if test.env != "" {
+				key, value, _ := strings.Cut(test.env, "=")
+				t.Setenv(key, value)
+			}
+			parent := t.TempDir()
+			missing := filepath.Join(parent, "rin-data")
+			arguments := append(
+				[]string{"serve", "-data", missing},
+				test.args...,
+			)
+			if err := run(arguments); err == nil {
+				t.Fatal("invalid limits started the Sidecar")
+			}
+			if _, err := os.Stat(missing); !errors.Is(err, os.ErrNotExist) {
+				t.Fatalf("invalid limits touched the data directory: %v", err)
+			}
+		})
+	}
+
+	clearServeEnvironment(t)
+	existing := t.TempDir()
+	sentinelPath := filepath.Join(existing, "operator-owned")
+	sentinel := []byte("must remain byte-identical")
+	if err := os.WriteFile(sentinelPath, sentinel, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := run([]string{
+		"serve",
+		"-data", existing,
+		"-transfer-max-events",
+		fmt.Sprint(uint64(protocol.MaxJSONSafeInteger) + 1),
+	}); err == nil {
+		t.Fatal("invalid Transfer event limit started the Sidecar")
+	}
+	after, err := os.ReadFile(sentinelPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(after, sentinel) {
+		t.Fatal("invalid configuration changed existing operator data")
+	}
+	entries, err := os.ReadDir(existing)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(entries) != 1 || entries[0].Name() != "operator-owned" {
+		t.Fatalf("invalid configuration changed existing directory: %v", entries)
+	}
+}
+
 func TestRunScrubLoopStartsImmediatelyAndStopsWithContext(t *testing.T) {
 	scrubber := &blockingScrubber{calls: make(chan int, 1)}
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))

docs/operations.md

@@ -105,7 +105,10 @@ are all present.
 
 Capacity, concurrency, timeout, and boolean environment variables fail fast
 when explicitly set to an invalid value; Rin does not silently replace a typo
-with a default. The same rule applies to explicit non-positive CLI limits.
+with a default. The same rule applies to explicit CLI limits. Runtime,
+Proposal Job, and Generation lower/upper bounds are all validated before Rin
+opens or performs recovery on the data directory, so a rejected configuration
+does not create or maintain Store files.
 
 The bundled Sidecar starts a checkpoint-independent event-log scrub
 immediately, then every 15 minutes. Each pass verifies at most 4,096 events and

docs/operations.zh-CN.md

@@ -95,7 +95,9 @@ rin serve -addr 10.0.0.12:7374 -allow-remote
 `-allow-remote`、Token 或该声明，会在打开数据目录前直接失败。
 
 容量、并发、Timeout 与 Boolean 环境变量一旦显式设置为非法值，Rin 会立即失败，
-不会把拼写错误静默替换成默认值；命令行显式设置的非正数 Limit 也遵循同一规则。
+不会把拼写错误静默替换成默认值；命令行显式 Limit 也遵循同一规则。Runtime、
+Proposal Job 与 Generation 的上下限全部在打开数据目录或执行恢复维护前完成校验，
+因此被拒绝的配置不会创建或维护 Store 文件。
 
 随附 Sidecar 会在启动后立即运行一次 checkpoint-independent Event Log Scrub，
 之后默认每 15 分钟运行一次。每个 Pass 最多校验 4,096 个事件，Deadline 为

generation/manager.go

@@ -130,23 +130,54 @@ func New(
 	if generationProvider == nil {
 		return nil, errors.New("structured generation provider is required")
 	}
+	config, err := normalizeConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	manager := &Manager{
+		provider: generationProvider, config: config, ctx: ctx, cancel: cancel,
+		queue: make(chan string, config.QueueSize), jobs: make(map[string]*jobState),
+		byRequest: make(map[string]string), cache: make(map[string]cacheEntry),
+		now: time.Now, done: make(chan struct{}),
+	}
+	for index := 0; index < config.Workers; index++ {
+		manager.wait.Add(1)
+		go manager.worker()
+	}
+	manager.wait.Add(1)
+	go manager.cleanupWorker()
+	go func() {
+		manager.wait.Wait()
+		close(manager.done)
+	}()
+	return manager, nil
+}
+
+// ValidateConfig applies the same limits as New without starting workers.
+func ValidateConfig(config Config) error {
+	_, err := normalizeConfig(config)
+	return err
+}
+
+func normalizeConfig(config Config) (Config, error) {
 	if config.Workers <= 0 {
 		config.Workers = 2
 	}
 	if config.Workers > 32 {
-		return nil, errors.New("generation workers must not exceed 32")
+		return Config{}, errors.New("generation workers must not exceed 32")
 	}
 	if config.QueueSize <= 0 {
 		config.QueueSize = 64
 	}
 	if config.QueueSize > 4096 {
-		return nil, errors.New("generation queue size must not exceed 4096")
+		return Config{}, errors.New("generation queue size must not exceed 4096")
 	}
 	if config.MaxJobs <= 0 {
 		config.MaxJobs = 512
 	}
 	if config.MaxJobs < config.QueueSize || config.MaxJobs > 16384 {
-		return nil, errors.New("generation max jobs must be between queue size and 16384")
+		return Config{}, errors.New("generation max jobs must be between queue size and 16384")
 	}
 	if config.JobTTL <= 0 {
 		config.JobTTL = 30 * time.Minute
@@ -155,7 +186,7 @@ func New(
 		config.CacheEntries = 256
 	}
 	if config.CacheEntries > 16384 {
-		return nil, errors.New("generation cache entries must not exceed 16384")
+		return Config{}, errors.New("generation cache entries must not exceed 16384")
 	}
 	if config.CacheTTL <= 0 {
 		config.CacheTTL = 30 * time.Minute
@@ -164,15 +195,17 @@ func New(
 		config.MaxOutputBytes = 512 * 1024
 	}
 	if config.MaxOutputBytes < 1024 || config.MaxOutputBytes > 4*1024*1024 {
-		return nil, errors.New("generation output limit must be between 1 KiB and 4 MiB")
+		return Config{}, errors.New(
+			"generation output limit must be between 1 KiB and 4 MiB",
+		)
 	}
 	if config.MaxRetainedBytes == 0 {
 		config.MaxRetainedBytes = 64 << 20
 	}
 	minimumRetained := uint64(config.MaxOutputBytes)*2 + (64 << 10)
 	if config.MaxRetainedBytes < minimumRetained ||
 		config.MaxRetainedBytes > 1<<30 {
-		return nil, errors.New(
+		return Config{}, errors.New(
 			"generation retained-memory limit must fit two outputs plus 64 KiB and not exceed 1 GiB",
 		)
 	}
@@ -181,29 +214,11 @@ func New(
 	}
 	if config.CleanupInterval < 10*time.Millisecond ||
 		config.CleanupInterval > time.Hour {
-		return nil, errors.New(
+		return Config{}, errors.New(
 			"generation cleanup interval must be between 10 ms and 1 hour",
 		)
 	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	manager := &Manager{
-		provider: generationProvider, config: config, ctx: ctx, cancel: cancel,
-		queue: make(chan string, config.QueueSize), jobs: make(map[string]*jobState),
-		byRequest: make(map[string]string), cache: make(map[string]cacheEntry),
-		now: time.Now, done: make(chan struct{}),
-	}
-	for index := 0; index < config.Workers; index++ {
-		manager.wait.Add(1)
-		go manager.worker()
-	}
-	manager.wait.Add(1)
-	go manager.cleanupWorker()
-	go func() {
-		manager.wait.Wait()
-		close(manager.done)
-	}()
-	return manager, nil
+	return config, nil
 }
 
 func (m *Manager) Submit(request protocol.GenerationRequest) (protocol.GenerationJobSubmission, error) {