From 2f0b93451cc9180d6790ca46fe550c2ab81b45ab Mon Sep 17 00:00:00 2001 From: "yunlu.wen" Date: Mon, 2 Nov 2026 12:43:48 +0800 Subject: [PATCH 1/3] feat(serverless): keep long-running invocations alive via periodic activation Long-running serverless invocations could be scaled to zero by the connector mid-flight. After a successful activation preflight, the daemon now re-calls the connector's activate endpoint every DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL seconds (default 60) until the invocation finishes. Keepalive failures are logged and never abort the invocation. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .env.example | 3 + docs/runtime/sri.md | 2 + docs/runtime/sri_cn.md | 2 + internal/core/serverless_runtime/io.go | 39 +++- internal/core/serverless_runtime/io_test.go | 201 ++++++++++++++++++++ internal/core/serverless_runtime/type.go | 24 ++- internal/types/app/config.go | 9 + internal/types/app/config_test.go | 20 ++ internal/types/app/default.go | 1 + 9 files changed, 290 insertions(+), 11 deletions(-) diff --git a/.env.example b/.env.example index b661f6f59..6dc4b1f95 100644 --- a/.env.example +++ b/.env.example @@ -173,6 +173,9 @@ DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_ENABLED=false # seconds to wait for the plugin to become ready during the activation preflight; # if it is not woken up in time the invocation is treated as failed DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_TIMEOUT=60 +# seconds between keepalive calls to the connector's activate endpoint while an +# invocation is in flight; the connector idle time should be at least 2x this interval +DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL=60 # python interpreter, if you are using local runtime, you should set this path to your python interpreter path # otherwise, it should be /usr/bin/python3 diff --git a/docs/runtime/sri.md b/docs/runtime/sri.md index 132928711..fb7438e77 100644 --- a/docs/runtime/sri.md +++ b/docs/runtime/sri.md @@ -19,6 +19,7 @@ The daemon is configured using the following environment variables: | `MAX_SERVERLESS_REQUEST_BYTES` | Maximum serialized request payload size sent to a serverless plugin runtime (in bytes). Default is 5242880 (5 MB). This limit accounts for Lambda Function URL's 6 MB request size limit, with a safety margin for headers and metadata. | | `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_ENABLED` | Whether to run the activation preflight before dispatching an invocation. When enabled, the daemon calls `POST /v1/activation/activate` to wake a scaled-to-zero plugin and waits until it is ready. Default is `false`. | | `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_TIMEOUT` | Seconds the daemon waits for the plugin to become ready during the activation preflight. If the plugin is not woken up in time, the invocation is treated as failed. Default is `60`. | +| `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL` | Seconds between keepalive calls to `POST /v1/activation/activate` while an invocation is in flight, preventing the plugin from being scaled down mid-invocation. Only effective when activation is enabled. The connector idle time should be at least 2x this interval. Keepalive failures are logged and do not fail the invocation. Default is `60`. | --- @@ -165,6 +166,7 @@ Content-Type: application/json **Error Handling** - The daemon bounds the wait with `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_TIMEOUT`. On timeout or any non-`200` response, the invocation is aborted and reported as a failure. +- After a successful preflight, the daemon re-calls this endpoint every `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL` seconds for the duration of the invocation to renew the activity lease. Keepalive failures are logged but never abort the invocation. --- diff --git a/docs/runtime/sri_cn.md b/docs/runtime/sri_cn.md index 24513e015..640e9599f 100644 --- a/docs/runtime/sri_cn.md +++ b/docs/runtime/sri_cn.md @@ -19,6 +19,7 @@ daemon 通过如下环境变量进行配置： | `MAX_SERVERLESS_REQUEST_BYTES` | 发送到 serverless 插件运行时的最大序列化请求负载大小（以字节为单位）。默认值为 5242880（5 MB）。此限制考虑了 Lambda Function URL 的 6 MB 请求大小限制，并为 headers 和元数据留有安全余量。 | | `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_ENABLED` | 是否在下发调用前执行激活（activation）预检。启用后，daemon 会调用 `POST /v1/activation/activate` 唤醒被缩容至零的插件，并等待其就绪。默认值为 `false`。 | | `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_TIMEOUT` | 激活预检中，daemon 等待插件就绪的超时时间（秒）。若插件未在超时时间内被唤醒，则本次调用按失败处理。默认值为 `60`。 | +| `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL` | 调用执行期间，daemon 周期性调用 `POST /v1/activation/activate` 保持插件活跃的间隔（秒），防止长时间运行的调用被缩容中断。仅在启用激活时生效。connector 的空闲时间应至少为该间隔的 2 倍。keepalive 失败仅记录日志，不会导致调用失败。默认值为 `60`。 | --- @@ -165,6 +166,7 @@ Content-Type: application/json **错误处理** - daemon 使用 `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_TIMEOUT` 限制等待时长。一旦超时或收到非 `200` 响应，本次调用将被中断并按失败处理。 +- 预检成功后，daemon 会在调用存续期间每隔 `DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL` 秒再次调用该接口以续期活跃租约。keepalive 失败仅记录日志，不会中断调用。 --- diff --git a/internal/core/serverless_runtime/io.go b/internal/core/serverless_runtime/io.go index edc9c6cf1..7ea282398 100644 --- a/internal/core/serverless_runtime/io.go +++ b/internal/core/serverless_runtime/io.go @@ -26,6 +26,8 @@ import ( const serverlessErrorResponseLimit = 4 * 1024 +var activatePlugin = serverless.Activate + // ServerlessPayloadTooLargeError reports request size metadata without exposing payload contents. type ServerlessPayloadTooLargeError struct { Action access_types.PluginAccessAction @@ -365,7 +367,7 @@ func (r *ServerlessPluginRuntime) WriteContext( }() if r.ActivationEnabled { - if err := serverless.Activate( + if err := activatePlugin( requestCtx, r.LambdaName, time.Duration(r.ActivationTimeout)*time.Second, @@ -387,6 +389,8 @@ func (r *ServerlessPluginRuntime) WriteContext( }) return } + stopKeepalive := r.startActivationKeepalive(requestCtx, sessionId, action) + defer stopKeepalive() } url += "?action=" + string(action) @@ -548,3 +552,36 @@ func (r *ServerlessPluginRuntime) WriteContext( return nil } + +func (r *ServerlessPluginRuntime) startActivationKeepalive(ctx context.Context, sessionID string, action access_types.PluginAccessAction) func() { + if r.ActivationKeepaliveInterval <= 0 { + return func() {} + } + ctx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(r.ActivationKeepaliveInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + if err := activatePlugin(ctx, r.LambdaName, r.ActivationKeepaliveInterval); err != nil && ctx.Err() == nil { + log.Warn( + "serverless runtime activation keepalive failed", + "session_id", sessionID, + "action", action, + "lambda_name", r.LambdaName, + "error", err, + ) + } + } + }() + return func() { + cancel() + <-done + } +} diff --git a/internal/core/serverless_runtime/io_test.go b/internal/core/serverless_runtime/io_test.go index b0b90e29e..a62e1fb2f 100644 --- a/internal/core/serverless_runtime/io_test.go +++ b/internal/core/serverless_runtime/io_test.go @@ -1062,3 +1062,204 @@ func (t *trackingReadCloser) Close() error { t.bodiesClosed.Add(1) return t.ReadCloser.Close() } + +func startWriteWithActivation( + t *testing.T, + handler http.HandlerFunc, + activationEnabled bool, + keepaliveInterval time.Duration, + activate func(ctx context.Context, instanceID string, timeout time.Duration) error, +) (chan plugin_entities.SessionMessage, chan struct{}) { + t.Helper() + routine.InitPool(1) + + originalActivate := activatePlugin + activatePlugin = activate + t.Cleanup(func() { activatePlugin = originalActivate }) + + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + + runtime := &ServerlessPluginRuntime{ + Client: server.Client(), + LambdaURL: server.URL, + LambdaName: "test-lambda", + MaxRequestBytes: 1024, + MaxRetryTimes: 1, + PluginMaxExecutionTimeout: 10, + RuntimeBufferSize: 1024, + RuntimeMaxBufferSize: 1024, + ActivationEnabled: activationEnabled, + ActivationTimeout: 5, + ActivationKeepaliveInterval: keepaliveInterval, + listeners: mapping.Map[string, *entities.Broadcast[plugin_entities.SessionMessage]]{}, + } + + listener, err := runtime.Listen("test-session") + if err != nil { + t.Fatalf("listen failed: %v", err) + } + + messages := make(chan plugin_entities.SessionMessage, 8) + closed := make(chan struct{}) + listener.Listen(func(message plugin_entities.SessionMessage) { + messages <- message + }) + listener.OnClose(func() { + close(closed) + }) + + if err := runtime.Write( + "test-session", + access_types.PLUGIN_ACCESS_ACTION_INVOKE_TEXT_EMBEDDING, + []byte("{}"), + ); err != nil { + t.Fatalf("write failed: %v", err) + } + + return messages, closed +} + +func collectMessagesAfterClose(t *testing.T, messages chan plugin_entities.SessionMessage, closed chan struct{}) []plugin_entities.SessionMessage { + t.Helper() + select { + case <-closed: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for listener close") + } + close(messages) + received := make([]plugin_entities.SessionMessage, 0, len(messages)) + for message := range messages { + received = append(received, message) + } + return received +} + +func slowEndSessionHandler(w http.ResponseWriter, r *http.Request) { + time.Sleep(150 * time.Millisecond) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte( + "{\"session_id\":\"test-session\",\"event\":\"session\",\"data\":{\"type\":\"end\",\"data\":{\"reason\":\"done\"}}}\n", + )) + w.(http.Flusher).Flush() +} + +func TestWrite_ActivationKeepaliveRenewsDuringLongInvocation(t *testing.T) { + var activateCalls atomic.Int32 + messages, closed := startWriteWithActivation( + t, + slowEndSessionHandler, + true, + 20*time.Millisecond, + func(ctx context.Context, instanceID string, timeout time.Duration) error { + activateCalls.Add(1) + return nil + }, + ) + + received := collectMessagesAfterClose(t, messages, closed) + + if calls := activateCalls.Load(); calls < 4 { + t.Fatalf("expected at least 4 activate calls (1 preflight + keepalives), got %d", calls) + } + hasEnd := false + for _, message := range received { + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_ERROR { + t.Fatalf("unexpected error message: %s", message.Data) + } + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_END { + hasEnd = true + } + } + if !hasEnd { + t.Fatal("expected END message") + } +} + +func TestWrite_ActivationKeepaliveStopsAfterInvocation(t *testing.T) { + var activateCalls atomic.Int32 + messages, closed := startWriteWithActivation( + t, + slowEndSessionHandler, + true, + 20*time.Millisecond, + func(ctx context.Context, instanceID string, timeout time.Duration) error { + activateCalls.Add(1) + return nil + }, + ) + + collectMessagesAfterClose(t, messages, closed) + + snapshot := activateCalls.Load() + time.Sleep(100 * time.Millisecond) + if calls := activateCalls.Load(); calls != snapshot { + t.Fatalf("expected no activate calls after invocation ended, got %d -> %d", snapshot, calls) + } +} + +func TestWrite_ActivationKeepaliveFailureDoesNotAbortInvocation(t *testing.T) { + var activateCalls atomic.Int32 + messages, closed := startWriteWithActivation( + t, + slowEndSessionHandler, + true, + 20*time.Millisecond, + func(ctx context.Context, instanceID string, timeout time.Duration) error { + if activateCalls.Add(1) == 1 { + return nil + } + return errors.New("activation keepalive failed") + }, + ) + + received := collectMessagesAfterClose(t, messages, closed) + + if calls := activateCalls.Load(); calls < 2 { + t.Fatalf("expected keepalive to be attempted, got %d activate calls", calls) + } + hasEnd := false + for _, message := range received { + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_ERROR { + t.Fatalf("keepalive failure must not produce an error message: %s", message.Data) + } + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_END { + hasEnd = true + } + } + if !hasEnd { + t.Fatal("expected END message") + } +} + +func TestWrite_ActivationDisabledSkipsKeepalive(t *testing.T) { + var activateCalls atomic.Int32 + messages, closed := startWriteWithActivation( + t, + slowEndSessionHandler, + false, + 20*time.Millisecond, + func(ctx context.Context, instanceID string, timeout time.Duration) error { + activateCalls.Add(1) + return nil + }, + ) + + received := collectMessagesAfterClose(t, messages, closed) + + if calls := activateCalls.Load(); calls != 0 { + t.Fatalf("expected no activate calls when activation is disabled, got %d", calls) + } + hasEnd := false + for _, message := range received { + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_ERROR { + t.Fatalf("unexpected error message: %s", message.Data) + } + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_END { + hasEnd = true + } + } + if !hasEnd { + t.Fatal("expected END message") + } +} diff --git a/internal/core/serverless_runtime/type.go b/internal/core/serverless_runtime/type.go index 74c889868..c23ad5cde 100644 --- a/internal/core/serverless_runtime/type.go +++ b/internal/core/serverless_runtime/type.go @@ -38,6 +38,9 @@ type ServerlessPluginRuntime struct { // ActivationTimeout bounds, in seconds, how long the preflight waits for the // plugin to become ready before treating the invocation as failed. ActivationTimeout int + // ActivationKeepaliveInterval is how often the daemon re-calls the + // connector's activate endpoint while an invocation is in flight. + ActivationKeepaliveInterval time.Duration RuntimeBufferSize int RuntimeMaxBufferSize int @@ -66,16 +69,17 @@ func ConstructServerlessPluginRuntime( MediaTransport: basic_runtime.NewMediaTransport(mediaBucket), InnerChecksum: serverlessModel.Checksum, }, - PluginRuntime: runtimeEntity, - LambdaURL: serverlessModel.FunctionURL, - LambdaName: serverlessModel.FunctionName, - PluginMaxExecutionTimeout: config.PluginMaxExecutionTimeout, - MaxRetryTimes: config.MaxServerlessRetryTimes, - MaxRequestBytes: config.MaxServerlessRequestBytes, - ActivationEnabled: config.DifyPluginServerlessConnectorActivationEnabled, - ActivationTimeout: config.DifyPluginServerlessConnectorActivationTimeout, - RuntimeBufferSize: config.PluginRuntimeBufferSize, - RuntimeMaxBufferSize: config.PluginRuntimeMaxBufferSize, + PluginRuntime: runtimeEntity, + LambdaURL: serverlessModel.FunctionURL, + LambdaName: serverlessModel.FunctionName, + PluginMaxExecutionTimeout: config.PluginMaxExecutionTimeout, + MaxRetryTimes: config.MaxServerlessRetryTimes, + MaxRequestBytes: config.MaxServerlessRequestBytes, + ActivationEnabled: config.DifyPluginServerlessConnectorActivationEnabled, + ActivationTimeout: config.DifyPluginServerlessConnectorActivationTimeout, + ActivationKeepaliveInterval: time.Duration(config.DifyPluginServerlessConnectorActivationKeepaliveInterval) * time.Second, + RuntimeBufferSize: config.PluginRuntimeBufferSize, + RuntimeMaxBufferSize: config.PluginRuntimeMaxBufferSize, // init http client Client: &http.Client{ diff --git a/internal/types/app/config.go b/internal/types/app/config.go index 703b65ec3..64935951d 100644 --- a/internal/types/app/config.go +++ b/internal/types/app/config.go @@ -218,6 +218,11 @@ type Config struct { // preflight. If the plugin is not woken up within this window the invocation // is treated as failed. DifyPluginServerlessConnectorActivationTimeout int `envconfig:"DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_TIMEOUT"` + // DifyPluginServerlessConnectorActivationKeepaliveInterval is the interval, in + // seconds, at which the daemon re-calls the connector's activate endpoint while + // an invocation is in flight, so a long-running invocation is not scaled down. + // The connector idle time should be at least twice this interval. + DifyPluginServerlessConnectorActivationKeepaliveInterval int `envconfig:"DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL"` MaxServerlessRetryTimes int `envconfig:"MAX_SERVERLESS_RETRY_TIMES" default:"3"` MaxServerlessRequestBytes int `envconfig:"MAX_SERVERLESS_REQUEST_BYTES" default:"5242880"` @@ -344,6 +349,10 @@ func (c *Config) Validate() error { if c.DifyPluginServerlessConnectorActivationEnabled && c.DifyPluginServerlessConnectorActivationTimeout <= 0 { return fmt.Errorf("dify plugin serverless connector activation timeout must be greater than zero when activation is enabled") } + + if c.DifyPluginServerlessConnectorActivationEnabled && c.DifyPluginServerlessConnectorActivationKeepaliveInterval <= 0 { + return fmt.Errorf("dify plugin serverless connector activation keepalive interval must be greater than zero when activation is enabled") + } case PLATFORM_LOCAL: if c.PluginWorkingPath == "" { return fmt.Errorf("plugin working path is empty") diff --git a/internal/types/app/config_test.go b/internal/types/app/config_test.go index eab69354a..61f2a449f 100644 --- a/internal/types/app/config_test.go +++ b/internal/types/app/config_test.go @@ -165,6 +165,26 @@ func TestValidateRejectsNonPositiveMaxServerlessRequestBytes(t *testing.T) { } } +func TestValidateRejectsNonPositiveActivationKeepaliveInterval(t *testing.T) { + for _, interval := range []int{0, -1} { + t.Run("keepalive_interval_"+strconv.Itoa(interval), func(t *testing.T) { + config := newValidConfigForValidation() + connectorURL := "http://localhost:5004" + connectorAPIKey := "test-key" + config.Platform = PLATFORM_SERVERLESS + config.DifyPluginServerlessConnectorURL = &connectorURL + config.DifyPluginServerlessConnectorAPIKey = &connectorAPIKey + config.DifyPluginServerlessConnectorActivationEnabled = true + config.DifyPluginServerlessConnectorActivationTimeout = 60 + config.DifyPluginServerlessConnectorActivationKeepaliveInterval = interval + + err := config.Validate() + + require.EqualError(t, err, "dify plugin serverless connector activation keepalive interval must be greater than zero when activation is enabled") + }) + } +} + func TestPipMirrorAutoDetectDefault(t *testing.T) { originalValue, hadOriginalValue := os.LookupEnv("PIP_MIRROR_AUTO_DETECT") require.NoError(t, os.Unsetenv("PIP_MIRROR_AUTO_DETECT")) diff --git a/internal/types/app/default.go b/internal/types/app/default.go index 4d4a87cd3..120e18d8d 100644 --- a/internal/types/app/default.go +++ b/internal/types/app/default.go @@ -32,6 +32,7 @@ func (config *Config) SetDefault() { setDefaultInt(&config.PluginAssetCacheSize, 256) setDefaultInt(&config.DifyPluginServerlessConnectorLaunchTimeout, 240) setDefaultInt(&config.DifyPluginServerlessConnectorActivationTimeout, 60) + setDefaultInt(&config.DifyPluginServerlessConnectorActivationKeepaliveInterval, 60) setDefaultInt(&config.PluginRemoteInstallingMaxSingleTenantConn, 5) setDefaultString(&config.DBSslMode, "disable") setDefaultString(&config.PluginStorageLocalRoot, "storage") From 3fac6afd857b4c49f7f01274466834e19d03e64c Mon Sep 17 00:00:00 2001 From: "yunlu.wen" Date: Mon, 2 Nov 2026 14:01:40 +0800 Subject: [PATCH 2/3] fix(serverless): recover panics in activation keepalive goroutine Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- internal/core/serverless_runtime/io.go | 8 ++++++ internal/core/serverless_runtime/io_test.go | 31 +++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/internal/core/serverless_runtime/io.go b/internal/core/serverless_runtime/io.go index 7ea282398..5fd8fbada 100644 --- a/internal/core/serverless_runtime/io.go +++ b/internal/core/serverless_runtime/io.go @@ -10,6 +10,7 @@ import ( "io" "net/http" "net/url" + "runtime" "strings" "time" @@ -561,6 +562,13 @@ func (r *ServerlessPluginRuntime) startActivationKeepalive(ctx context.Context, done := make(chan struct{}) go func() { defer close(done) + defer func() { + if err := recover(); err != nil { + buf := make([]byte, 1024*1024) + n := runtime.Stack(buf, false) + log.Error("panic in serverless activation keepalive", "session_id", sessionID, "action", action, "lambda_name", r.LambdaName, "panic", err, "stack_trace", string(buf[:n])) + } + }() ticker := time.NewTicker(r.ActivationKeepaliveInterval) defer ticker.Stop() for { diff --git a/internal/core/serverless_runtime/io_test.go b/internal/core/serverless_runtime/io_test.go index a62e1fb2f..62a5dc5ed 100644 --- a/internal/core/serverless_runtime/io_test.go +++ b/internal/core/serverless_runtime/io_test.go @@ -1263,3 +1263,34 @@ func TestWrite_ActivationDisabledSkipsKeepalive(t *testing.T) { t.Fatal("expected END message") } } + +func TestWrite_ActivationKeepalivePanicDoesNotCrashOrHang(t *testing.T) { + var activateCalls atomic.Int32 + messages, closed := startWriteWithActivation( + t, + slowEndSessionHandler, + true, + 20*time.Millisecond, + func(ctx context.Context, instanceID string, timeout time.Duration) error { + if activateCalls.Add(1) == 1 { + return nil + } + panic("keepalive panic") + }, + ) + + received := collectMessagesAfterClose(t, messages, closed) + + hasEnd := false + for _, message := range received { + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_ERROR { + t.Fatalf("unexpected error message: %s", message.Data) + } + if message.Type == plugin_entities.SESSION_MESSAGE_TYPE_END { + hasEnd = true + } + } + if !hasEnd { + t.Fatal("expected END message") + } +} From 0aec22ebf8254d9fe17e235bc94bb87fc2c32a3e Mon Sep 17 00:00:00 2001 From: "yunlu.wen" Date: Mon, 2 Nov 2026 14:34:57 +0800 Subject: [PATCH 3/3] fix(config): keep explicit zero keepalive interval so validation rejects it Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- internal/types/app/config.go | 2 +- internal/types/app/config_test.go | 59 +++++++++++++++++++++++++++++++ internal/types/app/default.go | 1 - 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/internal/types/app/config.go b/internal/types/app/config.go index 64935951d..b047721dc 100644 --- a/internal/types/app/config.go +++ b/internal/types/app/config.go @@ -222,7 +222,7 @@ type Config struct { // seconds, at which the daemon re-calls the connector's activate endpoint while // an invocation is in flight, so a long-running invocation is not scaled down. // The connector idle time should be at least twice this interval. - DifyPluginServerlessConnectorActivationKeepaliveInterval int `envconfig:"DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL"` + DifyPluginServerlessConnectorActivationKeepaliveInterval int `envconfig:"DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL" default:"60"` MaxServerlessRetryTimes int `envconfig:"MAX_SERVERLESS_RETRY_TIMES" default:"3"` MaxServerlessRequestBytes int `envconfig:"MAX_SERVERLESS_REQUEST_BYTES" default:"5242880"` diff --git a/internal/types/app/config_test.go b/internal/types/app/config_test.go index 61f2a449f..dbde86ec4 100644 --- a/internal/types/app/config_test.go +++ b/internal/types/app/config_test.go @@ -603,3 +603,62 @@ N8Qs9CZmKVFvJ4KU6nOaW5X8gTCrHvBFMFaQcKKpCmWZfLnPJZMdgZXvxhx5lXXU 9nKKqk7sKB4D6LqHKQ1qRx9HJJVP5LxHMYGpxGnxXNLMaCPjLMxVJpQSZLnVfj5Y dQ== -----END CERTIFICATE-----` + +func validateStartupServerlessActivationConfig(t *testing.T) (Config, error) { + t.Helper() + t.Setenv("PLATFORM", "serverless") + t.Setenv("SERVER_KEY", "test-key") + t.Setenv("DIFY_INNER_API_URL", "http://localhost") + t.Setenv("DIFY_INNER_API_KEY", "test-api-key") + t.Setenv("DB_USERNAME", "postgres") + t.Setenv("DB_PASSWORD", "postgres") + t.Setenv("DB_HOST", "localhost") + t.Setenv("DB_PORT", "5432") + t.Setenv("DB_DATABASE", "plugin") + t.Setenv("PLUGIN_REMOTE_INSTALLING_HOST", "localhost") + t.Setenv("PLUGIN_REMOTE_INSTALLING_PORT", "5003") + t.Setenv("DIFY_PLUGIN_SERVERLESS_CONNECTOR_URL", "http://localhost:5004") + t.Setenv("DIFY_PLUGIN_SERVERLESS_CONNECTOR_API_KEY", "test-key") + t.Setenv("DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_ENABLED", "true") + + var cfg Config + require.NoError(t, envconfig.Process("", &cfg)) + cfg.SetDefault() + return cfg, cfg.Validate() +} + +func TestActivationKeepaliveIntervalStartupDefault(t *testing.T) { + const keepaliveEnv = "DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL" + originalValue, hadOriginalValue := os.LookupEnv(keepaliveEnv) + require.NoError(t, os.Unsetenv(keepaliveEnv)) + t.Cleanup(func() { + if hadOriginalValue { + require.NoError(t, os.Setenv(keepaliveEnv, originalValue)) + return + } + require.NoError(t, os.Unsetenv(keepaliveEnv)) + }) + + cfg, err := validateStartupServerlessActivationConfig(t) + + require.NoError(t, err) + assert.Equal(t, 60, cfg.DifyPluginServerlessConnectorActivationKeepaliveInterval) +} + +func TestActivationKeepaliveIntervalStartupRejectsExplicitZero(t *testing.T) { + t.Setenv("DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL", "0") + + cfg, err := validateStartupServerlessActivationConfig(t) + + assert.Equal(t, 0, cfg.DifyPluginServerlessConnectorActivationKeepaliveInterval) + require.EqualError(t, err, "dify plugin serverless connector activation keepalive interval must be greater than zero when activation is enabled") +} + +func TestActivationKeepaliveIntervalStartupExplicitValue(t *testing.T) { + t.Setenv("DIFY_PLUGIN_SERVERLESS_CONNECTOR_ACTIVATION_KEEPALIVE_INTERVAL", "30") + + cfg, err := validateStartupServerlessActivationConfig(t) + + require.NoError(t, err) + assert.Equal(t, 30, cfg.DifyPluginServerlessConnectorActivationKeepaliveInterval) +} diff --git a/internal/types/app/default.go b/internal/types/app/default.go index 120e18d8d..4d4a87cd3 100644 --- a/internal/types/app/default.go +++ b/internal/types/app/default.go @@ -32,7 +32,6 @@ func (config *Config) SetDefault() { setDefaultInt(&config.PluginAssetCacheSize, 256) setDefaultInt(&config.DifyPluginServerlessConnectorLaunchTimeout, 240) setDefaultInt(&config.DifyPluginServerlessConnectorActivationTimeout, 60) - setDefaultInt(&config.DifyPluginServerlessConnectorActivationKeepaliveInterval, 60) setDefaultInt(&config.PluginRemoteInstallingMaxSingleTenantConn, 5) setDefaultString(&config.DBSslMode, "disable") setDefaultString(&config.PluginStorageLocalRoot, "storage")