This is an automated email from the ASF dual-hosted git repository.
nic-6443 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new 5e784af30d fix(ai-proxy): avoid aborting streams on empty flushes
(#13947)
5e784af30d is described below
commit 5e784af30d19ce508de5b88c5d438ddf69d156e0
Author: Nic <[email protected]>
AuthorDate: Tue Sep 15 17:52:11 2026 +0800
fix(ai-proxy): avoid aborting streams on empty flushes (#13947)
---
apisix/plugins/ai-providers/base.lua | 17 +-
docs/en/latest/plugins/ai-proxy.md | 2 +-
docs/zh/latest/plugins/ai-proxy.md | 2 +-
t/plugin/ai-proxy-flush-fragmented.t | 332 +++++++++++++++++++++++++++++++++++
4 files changed, 342 insertions(+), 11 deletions(-)
diff --git a/apisix/plugins/ai-providers/base.lua
b/apisix/plugins/ai-providers/base.lua
index 8612f5dfd6..63f74246ba 100644
--- a/apisix/plugins/ai-providers/base.lua
+++ b/apisix/plugins/ai-providers/base.lua
@@ -508,10 +508,9 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
local flush_interval_ms = conf and conf.streaming_flush_interval_ms or 0
-- async_flush: true when the interval thread is responsible for flushing
local async_flush = flush_interval_ms > 0
- -- needs_flush is set to true immediately after dispatching a chunk so the
- -- thread always flushes exactly the data that has been written. Cleared
- -- before ngx.flush() so any new chunks written during the flush yield are
- -- picked up on the next interval rather than silently dropped.
+ -- Arm after dispatching a chunk; a response filter may still withhold its
+ -- output, so the flush thread must tolerate "nothing to flush". Clear the
+ -- flag before flushing so a chunk dispatched during a yield arms it again.
local needs_flush = false
local flush_thread
local flush_err
@@ -524,11 +523,12 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
if needs_flush then
needs_flush = false
local ok, err = ngx.flush(false)
- if not ok then
+ if ok then
+ core.log.debug("ai-proxy: flush_thread periodic flush")
+ elseif err ~= "nothing to flush" then
flush_err = err
return
end
- core.log.debug("ai-proxy: flush_thread periodic flush")
end
end
end)
@@ -756,6 +756,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
return
end
output_sent = true
+ needs_flush = true
end
if ctx.var.llm_request_done and #converted_chunks == 0
@@ -766,6 +767,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
abort_on_disconnect(flush_err)
return
end
+ needs_flush = true
end
elseif ctx.ai_stream_framing ~= "sse" or complete ~= "" then
-- Native SSE filters need complete frames just like converters do.
@@ -778,9 +780,6 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
return
end
output_sent = true
- end
- -- Let the interval flush thread know there is unflushed output.
- if async_flush then
needs_flush = true
end
diff --git a/docs/en/latest/plugins/ai-proxy.md
b/docs/en/latest/plugins/ai-proxy.md
index 005f3108e4..5e24268e53 100644
--- a/docs/en/latest/plugins/ai-proxy.md
+++ b/docs/en/latest/plugins/ai-proxy.md
@@ -101,7 +101,7 @@ When `provider` is set to `bedrock`, the Plugin expects
requests in the [Bedrock
| keepalive_timeout | integer | False | 60000 | ≥ 1000
| Keepalive timeout in milliseconds when connecting to the LLM
service. |
| keepalive_pool | integer | False | 30 | ≥ 1
| Keepalive pool size for the LLM service connection. |
| ssl_verify | boolean | False | true |
| If true, verifies the LLM service's certificate. |
-| streaming_flush_interval_ms | integer | False | 10 | ≥ 0 | Interval in
milliseconds for the background flush thread. When `> 0` (default: `10`), a
background timer calls `ngx.flush(false)` every N ms, batching output for
bursty upstreams. When `0`, the background thread is disabled and each chunk is
flushed synchronously via `ngx.flush(true)`, guaranteeing immediate client
delivery. |
+| streaming_flush_interval_ms | integer | False | 10 | ≥ 0 | Interval in
milliseconds for the background flush thread. When `> 0` (default: `10`), a
background timer calls `ngx.flush(false)` every N ms, batching output for
bursty upstreams. For SSE streams, only complete frames or converted events
queued for the client trigger a flush; partial frames wait for the remaining
upstream data. If a periodic flush returns `nothing to flush`, the stream
continues. When `0`, the background thread [...]
## Provider-aware `max_tokens` mapping
diff --git a/docs/zh/latest/plugins/ai-proxy.md
b/docs/zh/latest/plugins/ai-proxy.md
index 62a1b4eae4..bdabdd300a 100644
--- a/docs/zh/latest/plugins/ai-proxy.md
+++ b/docs/zh/latest/plugins/ai-proxy.md
@@ -100,7 +100,7 @@ import TabItem from '@theme/TabItem';
| keepalive_timeout | integer | 否 | 60000 | ≥ 1000
| 连接到 LLM 服务时的保活超时时间（毫秒）。 |
| keepalive_pool | integer | 否 | 30 | ≥ 1
| LLM 服务连接的保活池大小。 |
| ssl_verify | boolean | 否 | true |
| 如果为 true，验证 LLM 服务的证书。 |
-| streaming_flush_interval_ms | integer | 否 | 10 | ≥ 0 | 后台刷新线程的间隔时间（毫秒）。`>
0`（默认值：`10`）时，后台定时器每隔 N 毫秒调用一次 `ngx.flush(false)`，适合上游批量发送 token 的场景。设为 `0`
时禁用后台线程，改为每个 chunk 同步调用 `ngx.flush(true)` 立即刷新。 |
+| streaming_flush_interval_ms | integer | 否 | 10 | ≥ 0 | 后台刷新线程的间隔时间（毫秒）。`>
0`（默认值：`10`）时，后台定时器每隔 N 毫秒调用一次 `ngx.flush(false)`，适合上游批量发送 token 的场景。对于 SSE
流，只有已排队发给客户端的完整帧或转换后的事件才会触发刷新；不完整的帧会等待后续上游数据。定时刷新若返回 `nothing to
flush`，会继续处理流式响应。设为 `0` 时禁用后台线程，改为每个 chunk 同步调用 `ngx.flush(true)` 立即刷新。 |
## Provider-aware `max_tokens` mapping
diff --git a/t/plugin/ai-proxy-flush-fragmented.t
b/t/plugin/ai-proxy-flush-fragmented.t
new file mode 100644
index 0000000000..c877970a1d
--- /dev/null
+++ b/t/plugin/ai-proxy-flush-fragmented.t
@@ -0,0 +1,332 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+use t::APISIX 'no_plan';
+use File::Slurp ();
+
+log_level("info");
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+ $block->set_value("extra_init_by_lua", <<'_EOC_');
+ local buffer_plugin = {
+ version = 0.1,
+ priority = 1,
+ name = "test-ai-buffer",
+ schema = {type = "object", properties = {}},
+ }
+ function buffer_plugin.check_schema(conf)
+ return require("apisix.core").schema.check(buffer_plugin.schema, conf)
+ end
+ function buffer_plugin.lua_body_filter(conf, ctx, headers, body)
+ ctx.test_ai_buffer = ctx.test_ai_buffer or {}
+ table.insert(ctx.test_ai_buffer, body)
+ if not ctx.var.llm_request_done then
+ return nil, ""
+ end
+ return nil, table.concat(ctx.test_ai_buffer)
+ end
+ package.loaded["apisix.plugins.test-ai-buffer"] = buffer_plugin
+_EOC_
+ $block->set_value("http_config", <<'_EOC_');
+ server {
+ listen 7752;
+ location /stream {
+ content_by_lua_block {
+ local loader = require("lib.fixture_loader")
+ local body =
assert(loader.load(ngx.req.get_headers()["X-AI-Fixture"]))
+ local args = ngx.req.get_uri_args()
+ ngx.header["Content-Type"] = "text/event-stream"
+ local function send(chunk)
+ if not ngx.print(chunk) or not ngx.flush(true) then
+ return false
+ end
+ ngx.sleep(0.05)
+ return true
+ end
+ if args.skip then
+ if not send(": keepalive\n\n") or not send(":
keepalive\n\n") then
+ return
+ end
+ end
+ if args.hold_eof then
+ ngx.shared.test:delete("converted-completion-received")
+ local done = assert(body:find("data: [DONE]", 1, true))
+ if not send(body:sub(1, done - 1)) then
+ return
+ end
+ assert(ngx.print(body:sub(done)))
+ assert(ngx.flush(true))
+ for _ = 1, 100 do
+ if
ngx.shared.test:get("converted-completion-received") then
+ return
+ end
+ ngx.sleep(0.01)
+ end
+ return
+ end
+ local offset = 0
+ local complete_frames = args.buffer and 2 or
args.after_first and 1 or 0
+ for _ = 1, complete_frames do
+ local boundary = assert(body:find("\n\n", offset + 1,
true)) + 1
+ if not send(body:sub(offset + 1, boundary)) then
+ return
+ end
+ offset = boundary
+ end
+ if args.buffer then
+ send(body:sub(offset + 1))
+ return
+ end
+ -- Two fragments leave the next frame incomplete across a
flush interval.
+ if not send(body:sub(offset + 1, offset + 10))
+ or not send(body:sub(offset + 11, offset + 20)) then
+ return
+ end
+ ngx.print(body:sub(offset + 21))
+ }
+ }
+ }
+_EOC_
+ $block->set_value("apisix_yaml", $block->apisix_yaml // <<'_EOC_');
+routes:
+ - id: native
+ uri: /fragmented/v1/responses
+ plugins:
+ ai-proxy-multi:
+ instances:
+ - name: native
+ provider: openai-compatible
+ weight: 1
+ auth:
+ header:
+ Authorization: Bearer test-key
+ override:
+ endpoint: http://127.0.0.1:7752/stream
+ - id: after-first
+ uri: /after-first/v1/responses
+ plugins:
+ ai-proxy-multi:
+ instances:
+ - name: native
+ provider: openai-compatible
+ weight: 1
+ auth:
+ header:
+ Authorization: Bearer test-key
+ override:
+ endpoint: http://127.0.0.1:7752/stream?after_first=true
+ - id: sync
+ uri: /sync/v1/responses
+ plugins:
+ ai-proxy-multi:
+ streaming_flush_interval_ms: 0
+ instances:
+ - name: native
+ provider: openai-compatible
+ weight: 1
+ auth:
+ header:
+ Authorization: Bearer test-key
+ override:
+ endpoint: http://127.0.0.1:7752/stream
+ - id: converted
+ uri: /converted/v1/messages
+ plugins:
+ ai-proxy:
+ provider: openai-compatible
+ auth:
+ header:
+ Authorization: Bearer test-key
+ override:
+ endpoint: http://127.0.0.1:7752/stream?skip=true
+#END
+_EOC_
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: async flush waits for the first complete native SSE frame
+--- request
+POST /fragmented/v1/responses
+{"model":"test","input":"hi","stream":true}
+--- more_headers
+X-AI-Fixture: openai/responses-streaming.sse
+--- response_body eval
+scalar File::Slurp::read_file("t/fixtures/openai/responses-streaming.sse")
+--- no_error_log
+[error]
+nothing to flush
+client disconnected during AI streaming
+
+
+
+=== TEST 2: async flush preserves partial frames after earlier output
+--- request
+POST /after-first/v1/responses
+{"model":"test","input":"hi","stream":true}
+--- more_headers
+X-AI-Fixture: openai/responses-streaming.sse
+--- response_body eval
+scalar File::Slurp::read_file("t/fixtures/openai/responses-streaming.sse")
+--- no_error_log
+[error]
+nothing to flush
+client disconnected during AI streaming
+
+
+
+=== TEST 3: synchronous flush waits for a complete native SSE frame
+--- request
+POST /sync/v1/responses
+{"model":"test","input":"hi","stream":true}
+--- more_headers
+X-AI-Fixture: openai/responses-streaming.sse
+--- response_body eval
+scalar File::Slurp::read_file("t/fixtures/openai/responses-streaming.sse")
+--- no_error_log
+[error]
+nothing to flush
+client disconnected during AI streaming
+
+
+
+=== TEST 4: converter waits through skipped events and partial frames before
output
+--- request
+POST /converted/v1/messages
+{"model":"test","messages":[{"role":"user","content":"hi"}],"max_tokens":32,"stream":true}
+--- more_headers
+X-AI-Fixture: protocol-conversion/openai-to-anthropic-stream.sse
+--- response_body_like eval
+qr/event: message_start.*Hello.* world.*event: message_stop/s
+--- no_error_log
+[error]
+nothing to flush
+client disconnected during AI streaming
+
+
+
+=== TEST 5: empty periodic flush resumes after a filter releases buffered
output
+--- extra_yaml_config
+plugins:
+ - ai-proxy-multi
+ - test-ai-buffer
+--- apisix_yaml
+routes:
+ - id: buffered
+ uri: /buffered/v1/responses
+ plugins:
+ test-ai-buffer: {}
+ ai-proxy-multi:
+ instances:
+ - name: native
+ provider: openai-compatible
+ weight: 1
+ auth:
+ header:
+ Authorization: Bearer test-key
+ override:
+ endpoint: http://127.0.0.1:7752/stream?buffer=true
+#END
+--- request
+POST /buffered/v1/responses
+{"model":"test","input":"hi","stream":true}
+--- more_headers
+X-AI-Fixture: openai/responses-streaming.sse
+--- response_body eval
+scalar File::Slurp::read_file("t/fixtures/openai/responses-streaming.sse")
+--- no_error_log
+[error]
+nothing to flush
+client disconnected during AI streaming
+
+
+
+=== TEST 6: empty converted completion dispatch flushes buffered output before
upstream EOF
+--- extra_yaml_config
+plugins:
+ - ai-proxy
+ - test-ai-buffer
+--- apisix_yaml
+routes:
+ - id: converted-buffered
+ uri: /converted-buffered/v1/messages
+ plugins:
+ test-ai-buffer: {}
+ ai-proxy:
+ provider: openai-compatible
+ auth:
+ header:
+ Authorization: Bearer test-key
+ override:
+ endpoint: http://127.0.0.1:7752/stream?hold_eof=true
+#END
+--- config
+ postpone_output 65536;
+ location /t {
+ content_by_lua_block {
+ local httpc = require("resty.http").new()
+ httpc:set_timeout(500)
+ assert(httpc:connect("127.0.0.1", ngx.var.server_port))
+ local res, err = httpc:request({
+ method = "POST",
+ path = "/converted-buffered/v1/messages",
+ headers = {
+ ["Content-Type"] = "application/json",
+ ["X-AI-Fixture"] =
"protocol-conversion/usage-only-final-chunk.sse",
+ },
+ body =
[[{"model":"test","messages":[{"role":"user","content":"hi"}],
+ "max_tokens":32,"stream":true}]],
+ })
+ local body = ""
+ if res then
+ while true do
+ local chunk
+ chunk, err = res.body_reader()
+ if not chunk then
+ break
+ end
+ body = body .. chunk
+ if body:find("event: message_stop", 1, true) then
+
assert(ngx.shared.test:set("converted-completion-received", true))
+ end
+ end
+ end
+ httpc:close()
+ if err then
+ assert(ngx.shared.test:set("converted-completion-received",
true))
+ ngx.say("failed: ", err)
+ return
+ end
+ assert(body:find('"text":"Hi"', 1, true), body)
+ assert(body:find("event: message_stop", 1, true), body)
+ ngx.say("complete response received before upstream EOF")
+ }
+ }
+--- request
+GET /t
+--- response_body
+complete response received before upstream EOF
+--- no_error_log
+[error]
+nothing to flush
+client disconnected during AI streaming
