feat: make turbo mode configurable via config file

Add TurboModeEnable setting to server configuration with environment variable override.
The DTAIL_TURBOBOOST_ENABLE environment variable takes precedence over config file setting.
Turbo mode is automatically disabled for MapReduce operations to prevent data accuracy issues.

- Add TurboModeEnable boolean to ServerConfig struct
- Update config initializer to check environment variable for backward compatibility
- Replace direct env var checks with config.Server.TurboModeEnable throughout codebase
- Enable turbo mode in example config file (dtail.json.example)
- Add property to JSON schema with descriptive documentation

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

6 files changed, 20 insertions, 6 deletions

diff --git a/examples/dtail.json.example b/examples/dtail.json.example
index 26eb8a1..f21d114 100644
--- a/examples/dtail.json.example
+++ b/examples/dtail.json.example
@@ -97,6 +97,7 @@
     "MaxConcurrentTails": 50,
     "MaxConnections": 50,
     "MaxLineLength": 1048576,
+    "TurboModeEnable": true,
     "Permissions": {
       "Default": [
         "readfiles:^/.*$"
diff --git a/examples/dtail.schema.json b/examples/dtail.schema.json
index d13b133..5690ad7 100755
--- a/examples/dtail.schema.json
+++ b/examples/dtail.schema.json
@@ -395,6 +395,10 @@
           "minimum": 1024,
           "maximum": 10240000
         },
+        "TurboModeEnable": {
+          "type": "boolean",
+          "description": "Enable turbo mode for optimized file processing (disabled by default for MapReduce)"
+        },
         "Permissions": {
           "type": "object",
           "additionalProperties": true,
diff --git a/internal/config/initializer.go b/internal/config/initializer.go
index 9c3bf64..e750a1a 100644
--- a/internal/config/initializer.go
+++ b/internal/config/initializer.go
@@ -92,6 +92,11 @@ func (in *initializer) processEnvVars(args *Args) {
 	if len(sshPrivateKeyPathFile) > 0 && args.SSHPrivateKeyFilePath == "" {
 		args.SSHPrivateKeyFilePath = sshPrivateKeyPathFile
 	}
+	// Check if turbo mode should be enabled from environment variable
+	// This allows backward compatibility with existing scripts
+	if Env("DTAIL_TURBOBOOST_ENABLE") {
+		in.Server.TurboModeEnable = true
+	}
 }
 
 func (in *initializer) setupConfig(sourceCb transformCb, args *Args,
diff --git a/internal/config/server.go b/internal/config/server.go
index cb9ca2b..d108ea3 100644
--- a/internal/config/server.go
+++ b/internal/config/server.go
@@ -67,6 +67,11 @@ type ServerConfig struct {
 	Ciphers []string `json:",omitempty"`
 	// The allowed MAC algorithms.
 	MACs []string `json:",omitempty"`
+	// Enable turbo mode for optimized file processing. When enabled, cat/grep/tail operations
+	// use a direct writing approach that bypasses internal channels for better performance.
+	// Note: This is disabled by default for MapReduce operations due to known issues with
+	// high-concurrency aggregate processing.
+	TurboModeEnable bool `json:",omitempty"`
 }
 
 // Create a new default server configuration.
@@ -85,6 +90,7 @@ func newDefaultServerConfig() *ServerConfig {
 		Permissions: Permissions{
 			Default: defaultPermissions,
 		},
+		TurboModeEnable:    false, // Default to false for safety, can be enabled via config
 	}
 }
 
diff --git a/internal/mapr/server/aggregate.go b/internal/mapr/server/aggregate.go
index 1f735ac..f055b9d 100644
--- a/internal/mapr/server/aggregate.go
+++ b/internal/mapr/server/aggregate.go
@@ -141,7 +141,7 @@ func (a *Aggregate) nextLine() (line *line.Line, ok bool, noMoreChannels bool) {
 			a.linesCh = newLinesCh
 			
 			// In turbo mode, synchronously put the channel back to avoid race conditions
-			if config.Env("DTAIL_TURBOBOOST_ENABLE") {
+			if config.Server.TurboModeEnable {
 				select {
 				case a.NextLinesCh <- oldLinesCh:
 					// Successfully put back
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 1f46498..bdb7b8b 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -121,8 +121,7 @@ func (r *readCommand) readFiles(ctx context.Context, ltx lcontext.LContext,
 
 	// In turbo mode, signal EOF after all files are processed
 	// This is crucial for proper shutdown in server mode
-	turboBoostEnabled := config.Env("DTAIL_TURBOBOOST_ENABLE")
-	if turboBoostEnabled && r.server.aggregate == nil && 
+	if config.Server.TurboModeEnable && r.server.aggregate == nil && 
 		(r.mode == omode.CatClient || r.mode == omode.GrepClient || r.mode == omode.TailClient) {
 		if r.server.IsTurboMode() && r.server.turboEOF != nil {
 			dlog.Server.Debug(r.server.user, "Turbo mode: flushing data before EOF signal")
@@ -249,10 +248,9 @@ func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext,
 	}
 
 	// Check if we should use the turbo boost optimizations
-	turboBoostEnabled := config.Env("DTAIL_TURBOBOOST_ENABLE")
 	// Enable turbo boost for cat/grep/tail modes, but NOT for aggregate (MapReduce) operations
 	// MapReduce requires the traditional channel-based approach to work correctly
-	if turboBoostEnabled && r.server.aggregate == nil &&
+	if config.Server.TurboModeEnable && r.server.aggregate == nil &&
 		(r.mode == omode.CatClient || r.mode == omode.GrepClient || r.mode == omode.TailClient) {
 		r.readWithTurboProcessor(ctx, ltx, path, globID, re, reader)
 		return
@@ -309,7 +307,7 @@ func (r *readCommand) readWithProcessor(ctx context.Context, ltx lcontext.LConte
 	aggregate := r.server.aggregate
 
 	// Use the optimized version if turbo boost is enabled
-	turboBoostEnabled := config.Env("DTAIL_TURBOBOOST_ENABLE")
+	turboBoostEnabled := config.Server.TurboModeEnable
 
 	for {
 		if aggregate != nil {