fix(plugin): support customized stdio buffer size, align serverless runtime with local runtime (#470)

* fix(plugin): support customized stdio buffer size, align serverless runtime with local runtime

* fix(plugin): update config naming to pluginRuntimeBufferSize & pluginRuntimeMaxBufferSize, and keep the compatibility to stdoutBufferSize

* fix: provide a default value for PLUGIN_RUNTIME_BUFFER_SIZE & PLUGIN_RUNTIME_MAX_BUFFER_SIZE

* fix: remove redundant logic

* fix: add compatibility to PLUGIN_STDIO_BUFFER_SIZE & PLUGIN_STDIO_MAX_BUFFER_SIZE

Author: TonloOO

.env.example

@@ -149,10 +149,14 @@ ENFORCE_LANGGENIUS_PLUGIN_SIGNATURES=true
 HTTP_PROXY=
 HTTPS_PROXY=
 
-# plugin stdio buffer size
+# plugin stdio buffer size(local runtime, will be deprecated in future version)
 PLUGIN_STDIO_BUFFER_SIZE=1024
 PLUGIN_STDIO_MAX_BUFFER_SIZE=5242880
 
+# plugin runtime buffer size(serverless & local runtime)
+PLUGIN_RUNTIME_BUFFER_SIZE=1024
+PLUGIN_RUNTIME_MAX_BUFFER_SIZE=5242880
+
 # dify backwards invocation write timeout in milliseconds
 DIFY_BACKWARDS_INVOCATION_WRITE_TIMEOUT=5000
 # dify backwards invocation read timeout in milliseconds

internal/core/plugin_manager/launcher.go

@@ -140,8 +140,8 @@ func (p *PluginManager) launchLocal(pluginUniqueIdentifier plugin_entities.Plugi
 		PipMirrorUrl:              p.config.PipMirrorUrl,
 		PipPreferBinary:           *p.config.PipPreferBinary,
 		PipExtraArgs:              p.config.PipExtraArgs,
-		StdoutBufferSize:          p.config.PluginStdioBufferSize,
-		StdoutMaxBufferSize:       p.config.PluginStdioMaxBufferSize,
+		StdoutBufferSize:          p.config.GetLocalRuntimeBufferSize(),
+		StdoutMaxBufferSize:       p.config.GetLocalRuntimeMaxBufferSize(),
 	})
 	localPluginRuntime.PluginRuntime = plugin.runtime
 	localPluginRuntime.BasicChecksum = basic_runtime.BasicChecksum{

internal/core/plugin_manager/serverless.go

@@ -44,7 +44,7 @@ func (p *PluginManager) getServerlessPluginRuntime(
 	runtimeEntity.InitState()
 
 	// convert to plugin runtime
-	pluginRuntime := serverless_runtime.ServerlessPluginRuntime{
+	pluginRuntime := &serverless_runtime.ServerlessPluginRuntime{
 		BasicChecksum: basic_runtime.BasicChecksum{
 			MediaTransport: basic_runtime.NewMediaTransport(p.mediaBucket),
 			InnerChecksum:  model.Checksum,
@@ -53,13 +53,15 @@ func (p *PluginManager) getServerlessPluginRuntime(
 		LambdaURL:                 model.FunctionURL,
 		LambdaName:                model.FunctionName,
 		PluginMaxExecutionTimeout: p.config.PluginMaxExecutionTimeout,
+		RuntimeBufferSize:         p.config.PluginRuntimeBufferSize,
+		RuntimeMaxBufferSize:      p.config.PluginRuntimeMaxBufferSize,
 	}
 
 	if err := pluginRuntime.InitEnvironment(); err != nil {
 		return nil, err
 	}
 
-	return &pluginRuntime, nil
+	return pluginRuntime, nil
 }
 
 func (p *PluginManager) getServerlessPluginRuntimeModel(

internal/core/plugin_manager/serverless_runtime/io.go

@@ -87,8 +87,7 @@ func (r *ServerlessPluginRuntime) Write(sessionId string, action access_types.Pl
 		scanner := bufio.NewScanner(response.Body)
 		defer response.Body.Close()
 
-		// TODO: set a reasonable buffer size or use a reader, this is a temporary solution
-		scanner.Buffer(make([]byte, 1024), 5*1024*1024)
+		scanner.Buffer(make([]byte, r.RuntimeBufferSize), r.RuntimeMaxBufferSize)
 
 		sessionAlive := true
 		for scanner.Scan() && sessionAlive {

internal/core/plugin_manager/serverless_runtime/type.go

@@ -23,4 +23,7 @@ type ServerlessPluginRuntime struct {
 	client *http.Client
 
 	PluginMaxExecutionTimeout int // in seconds
+
+	RuntimeBufferSize    int
+	RuntimeMaxBufferSize int
 }

internal/types/app/config.go

@@ -173,6 +173,13 @@ type Config struct {
 	PipVerbose                *bool  `envconfig:"PIP_VERBOSE"`
 	PipExtraArgs              string `envconfig:"PIP_EXTRA_ARGS"`
 
+	// Runtime buffer configuration (applies to both local and serverless runtimes)
+	// These are the new generic names that should be used going forward
+	PluginRuntimeBufferSize    int `envconfig:"PLUGIN_RUNTIME_BUFFER_SIZE" default:"1024"`
+	PluginRuntimeMaxBufferSize int `envconfig:"PLUGIN_RUNTIME_MAX_BUFFER_SIZE" default:"5242880"`
+
+	// Legacy STDIO-specific buffer configuration (kept for backward compatibility)
+	// If the new PluginRuntime* configs are not set, these will be used as fallback
 	PluginStdioBufferSize    int `envconfig:"PLUGIN_STDIO_BUFFER_SIZE" default:"1024"`
 	PluginStdioMaxBufferSize int `envconfig:"PLUGIN_STDIO_MAX_BUFFER_SIZE" default:"5242880"`
 
@@ -253,6 +260,22 @@ func (c *Config) Validate() error {
 	return nil
 }
 
+// Prefers Stdio (legacy) config if user has customized it, falls back to Runtime (new) config.
+func (c *Config) GetLocalRuntimeBufferSize() int {
+	if c.PluginStdioBufferSize != 1024 {
+		return c.PluginStdioBufferSize
+	}
+	return c.PluginRuntimeBufferSize
+}
+
+// Prefers Stdio (legacy) config if user has customized it, falls back to Runtime (new) config.
+func (c *Config) GetLocalRuntimeMaxBufferSize() int {
+	if c.PluginStdioMaxBufferSize != 5242880 {
+		return c.PluginStdioMaxBufferSize
+	}
+	return c.PluginRuntimeMaxBufferSize
+}
+
 type PlatformType string
 
 const (