This is an automated email from the ASF dual-hosted git repository.
shreemaan-abhishek 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 c3141d49c0 fix(batch-requests): bound aggregated response bodies
(#13906)
c3141d49c0 is described below
commit c3141d49c01fc704ba3baba8892427806b636160
Author: Shreemaan Abhishek <[email protected]>
AuthorDate: Fri Sep 11 18:34:28 2026 +0800
fix(batch-requests): bound aggregated response bodies (#13906)
---
apisix-master-0.rockspec | 2 +-
apisix/plugins/batch-requests.lua | 102 ++++++++++++++-
t/plugin/batch-requests.t | 266 ++++++++++++++++++++++++++++++++++++++
3 files changed, 366 insertions(+), 4 deletions(-)
diff --git a/apisix-master-0.rockspec b/apisix-master-0.rockspec
index 064aaea905..9a6a9fa9d9 100644
--- a/apisix-master-0.rockspec
+++ b/apisix-master-0.rockspec
@@ -37,7 +37,7 @@ dependencies = {
"api7-lua-resty-dns-client = 7.1.2-0",
"lua-resty-template = 2.0-1",
"lua-resty-etcd = 1.10.6-0",
- "api7-lua-resty-http = 0.2.3-0",
+ "api7-lua-resty-http = 0.2.4-0",
"lua-resty-balancer = 0.05-0",
"lua-resty-ngxvar = 0.5.2-0",
"lua-resty-jit-uuid = 0.0.7-2",
diff --git a/apisix/plugins/batch-requests.lua
b/apisix/plugins/batch-requests.lua
index 8927b33b5a..d8faa288b6 100644
--- a/apisix/plugins/batch-requests.lua
+++ b/apisix/plugins/batch-requests.lua
@@ -20,6 +20,7 @@ local plugin = require("apisix.plugin")
local ngx = ngx
local ipairs = ipairs
local pairs = pairs
+local tonumber = tonumber
local str_find = core.string.find
local str_lower = string.lower
@@ -45,6 +46,9 @@ local schema = {
local default_max_body_size = 1024 * 1024 -- 1MiB
local default_max_pipeline_items = 1000
+local default_max_response_body_size = 1024 * 1024 -- 1MiB
+local default_max_response_body_size_total = 10 * 1024 * 1024 -- 10MiB
+local response_body_chunk_size = 8192
local metadata_schema = {
type = "object",
properties = {
@@ -60,6 +64,18 @@ local metadata_schema = {
exclusiveMinimum = 0,
default = default_max_pipeline_items,
},
+ max_response_body_size = {
+ description = "max response body size in bytes for each pipeline
request",
+ type = "integer",
+ exclusiveMinimum = 0,
+ default = default_max_response_body_size,
+ },
+ max_response_body_size_total = {
+ description = "max total response body size in bytes for a
pipeline",
+ type = "integer",
+ exclusiveMinimum = 0,
+ default = default_max_response_body_size_total,
+ },
},
}
@@ -224,18 +240,87 @@ local function set_common_query(data)
end
+local function close_http_client(httpc)
+ local ok, err = httpc:close()
+ if not ok then
+ core.log.warn("failed to close batch request connection: ", err)
+ end
+end
+
+
+local function read_response_body(httpc, resp, max_response_body_size,
+ response_body_size_total,
+ max_response_body_size_total)
+ local content_length = tonumber(resp.headers["Content-Length"])
+ local close_delimited = not content_length and
+ not http.transfer_encoding_is_chunked(resp.headers)
+ if content_length then
+ if content_length > max_response_body_size then
+ close_http_client(httpc)
+ return nil, nil, "max_response_body_size"
+ end
+
+ if response_body_size_total + content_length >
max_response_body_size_total then
+ close_http_client(httpc)
+ return nil, nil, "max_response_body_size_total"
+ end
+ end
+
+ local chunks = {}
+ local response_body_size = 0
+ while true do
+ local chunk, err = resp.body_reader(response_body_chunk_size)
+ local valid_eof = close_delimited and err == "closed"
+ if err and not valid_eof then
+ close_http_client(httpc)
+ return nil, err
+ end
+
+ if chunk then
+ response_body_size = response_body_size + #chunk
+ if response_body_size > max_response_body_size then
+ close_http_client(httpc)
+ return nil, nil, "max_response_body_size"
+ end
+
+ if response_body_size_total + response_body_size >
+ max_response_body_size_total then
+ close_http_client(httpc)
+ return nil, nil, "max_response_body_size_total"
+ end
+
+ core.table.insert(chunks, chunk)
+ end
+
+ if valid_eof or not chunk then
+ break
+ end
+ end
+
+ return core.table.concat(chunks), nil, nil, response_body_size
+end
+
+
local function batch_requests(ctx)
local metadata = plugin.plugin_metadata(plugin_name)
core.log.info("metadata: ", core.json.delay_encode(metadata))
local max_body_size
local max_pipeline_items
+ local max_response_body_size
+ local max_response_body_size_total
if metadata then
max_body_size = metadata.value.max_body_size
max_pipeline_items = metadata.value.max_pipeline_items or
default_max_pipeline_items
+ max_response_body_size = metadata.value.max_response_body_size or
+ default_max_response_body_size
+ max_response_body_size_total =
metadata.value.max_response_body_size_total or
+ default_max_response_body_size_total
else
max_body_size = default_max_body_size
max_pipeline_items = default_max_pipeline_items
+ max_response_body_size = default_max_response_body_size
+ max_response_body_size_total = default_max_response_body_size_total
end
local req_body, err = core.request.get_body(max_body_size, ctx)
@@ -286,7 +371,8 @@ local function batch_requests(ctx)
end
local aggregated_resp = {}
- for _, resp in ipairs(responses) do
+ local response_body_size_total = 0
+ for i, resp in ipairs(responses) do
if not resp.status then
core.table.insert(aggregated_resp, {
status = 504,
@@ -300,12 +386,22 @@ local function batch_requests(ctx)
headers = resp.headers,
}
if resp.has_body then
- local err
- sub_resp.body, err = resp:read_body()
+ local err, limit_name, response_body_size
+ sub_resp.body, err, limit_name, response_body_size =
+ read_response_body(httpc, resp, max_response_body_size,
+ response_body_size_total,
+ max_response_body_size_total)
+ if limit_name then
+ return 502, {
+ error_msg = "response body of pipeline request " .. i ..
+ " exceeds " .. limit_name
+ }
+ end
if err then
sub_resp.read_body_err = err
core.log.error("read pipeline response body failed: ", err)
else
+ response_body_size_total = response_body_size_total +
response_body_size
resp:read_trailers()
end
end
diff --git a/t/plugin/batch-requests.t b/t/plugin/batch-requests.t
index 298f9df11e..ed6e1e8f92 100644
--- a/t/plugin/batch-requests.t
+++ b/t/plugin/batch-requests.t
@@ -1194,3 +1194,269 @@ qr/property \\"path\\" is required/
GET /t
--- response_body
passed
+
+
+
+=== TEST 31: configure response body limits
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body =
t('/apisix/admin/plugin_metadata/batch-requests',
+ ngx.HTTP_PUT,
+ [[{
+ "max_response_body_size": 5,
+ "max_response_body_size_total": 8
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 32: reject a subresponse body over its limit
+--- config
+ location = /six-bytes {
+ content_by_lua_block {
+ ngx.header.content_length = 6
+ ngx.print("123456")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"pipeline":[{"path":"/six-bytes"}]}
+--- error_code: 502
+--- response_body
+{"error_msg":"response body of pipeline request 1 exceeds
max_response_body_size"}
+
+
+
+=== TEST 33: reject aggregate response bodies over their limit
+--- config
+ location = /five-bytes {
+ content_by_lua_block {
+ ngx.print("12")
+ ngx.flush(true)
+ ngx.print("345")
+ }
+ }
+ location = /four-bytes {
+ content_by_lua_block {
+ ngx.print("12")
+ ngx.flush(true)
+ ngx.print("34")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"headers":{"Connection":"keep-alive"},"pipeline":[{"path":"/five-bytes"},{"path":"/four-bytes"}]}
+--- error_code: 502
+--- response_body
+{"error_msg":"response body of pipeline request 2 exceeds
max_response_body_size_total"}
+
+
+
+=== TEST 34: allow response bodies at the per-item and aggregate limits
+--- config
+ location = /five-bytes {
+ content_by_lua_block {
+ ngx.print("12")
+ ngx.flush(true)
+ ngx.print("345")
+ }
+ }
+ location = /three-bytes {
+ content_by_lua_block {
+ ngx.print("1")
+ ngx.flush(true)
+ ngx.print("23")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"headers":{"Connection":"keep-alive"},"pipeline":[{"path":"/five-bytes"},{"path":"/three-bytes"}]}
+--- error_code: 200
+--- response_body eval
+qr/^\[.*"body":"12345".*"body":"123".*\]$/
+
+
+
+=== TEST 35: accept a close-delimited response at the per-item limit
+--- config
+ location = /close-delimited-five-bytes {
+ chunked_transfer_encoding off;
+ content_by_lua_block {
+ ngx.print("12345")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"pipeline":[{"path":"/close-delimited-five-bytes","headers":{"Connection":"close"}}]}
+--- error_code: 200
+--- response_body eval
+qr/^\[.*"body":"12345".*\]$/
+--- no_error_log
+read pipeline response body failed
+
+
+
+=== TEST 36: reject a close-delimited response over the per-item limit
+--- config
+ location = /close-delimited-six-bytes {
+ chunked_transfer_encoding off;
+ content_by_lua_block {
+ ngx.print("123456")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"pipeline":[{"path":"/close-delimited-six-bytes","headers":{"Connection":"close"}}]}
+--- error_code: 502
+--- response_body
+{"error_msg":"response body of pipeline request 1 exceeds
max_response_body_size"}
+
+
+
+=== TEST 37: count a close-delimited response toward the aggregate limit
+--- config
+ location = /four-bytes-with-length {
+ content_by_lua_block {
+ ngx.header.content_length = 4
+ ngx.print("1234")
+ }
+ }
+ location = /close-delimited-five-bytes {
+ chunked_transfer_encoding off;
+ content_by_lua_block {
+ ngx.print("12345")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"headers":{"Connection":"keep-alive"},"pipeline":[{"path":"/four-bytes-with-length"},{"path":"/close-delimited-five-bytes","headers":{"Connection":"close"}}]}
+--- error_code: 502
+--- response_body
+{"error_msg":"response body of pipeline request 2 exceeds
max_response_body_size_total"}
+
+
+
+=== TEST 38: report a timeout while reading a close-delimited response
+--- config
+ location = /close-delimited-timeout {
+ chunked_transfer_encoding off;
+ content_by_lua_block {
+ ngx.print("partial")
+ ngx.flush(true)
+ ngx.sleep(0.3)
+ ngx.print("body")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"timeout":100,"pipeline":[{"path":"/close-delimited-timeout","headers":{"Connection":"close"}}]}
+--- error_code: 200
+--- response_body_like eval
+qr/"read_body_err":"timeout"/
+--- response_body_unlike eval
+qr/"body":/
+--- error_log
+read pipeline response body failed: timeout
+
+
+
+=== TEST 39: reject an invalid response body limit
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body =
t('/apisix/admin/plugin_metadata/batch-requests',
+ ngx.HTTP_PUT,
+ [[{
+ "max_response_body_size_total": 0
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- error_code: 400
+--- response_body eval
+qr/property \\"max_response_body_size_total\\" validation failed/
+
+
+
+=== TEST 40: reset plugin metadata
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body =
t('/apisix/admin/plugin_metadata/batch-requests',
+ ngx.HTTP_PUT,
+ [[{
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 41: reject a response body over the default per-item limit
+--- config
+ location = /over-one-mib {
+ content_by_lua_block {
+ ngx.header.content_length = 1048577
+ ngx.print(("x"):rep(1048577))
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"pipeline":[{"path":"/over-one-mib"}]}
+--- error_code: 502
+--- response_body
+{"error_msg":"response body of pipeline request 1 exceeds
max_response_body_size"}
+
+
+
+=== TEST 42: accept the default boundaries and reject the next aggregate byte
+--- config
+ location = /one-mib {
+ content_by_lua_block {
+ ngx.header.content_length = 1048576
+ ngx.print(("x"):rep(1048576))
+ }
+ }
+ location = /one-byte {
+ content_by_lua_block {
+ ngx.header.content_length = 1
+ ngx.print("x")
+ }
+ }
+--- request
+POST /apisix/batch-requests
+{"headers":{"Connection":"keep-alive"},"pipeline":[{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-mib"},{"path":"/one-byte"}]}
+--- error_code: 502
+--- response_body
+{"error_msg":"response body of pipeline request 11 exceeds
max_response_body_size_total"}
