fix: drain final response buffers before completing

src/WebResponseImpl.h

@@ -82,6 +82,8 @@ class AsyncAbstractResponse : public AsyncWebServerResponse {
   std::unique_ptr<std::array<uint8_t, ASYNC_RESPONCE_BUFF_SIZE> > _send_buffer;
   // buffer data size specifiers
   size_t _send_buffer_offset{0}, _send_buffer_len{0};
+  // Source EOF does not mean that the final buffer has reached the TCP queue.
+  bool _source_ended{false};
   size_t _readDataFromCacheOrContent(uint8_t *data, const size_t len);
   size_t _fillBufferAndProcessTemplates(uint8_t *buf, size_t maxLen);
 

src/WebResponses.cpp

@@ -351,6 +351,14 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
     return 0;
   }
 
+  _ackedLength += len;
+  if (_state == RESPONSE_WAIT_ACK) {
+    if (_ackedLength >= _writtenLength) {
+      _state = RESPONSE_END;
+    }
+    return 0;
+  }
+
 #if ASYNCWEBSERVER_USE_CHUNK_INFLIGHT
   /*
     for response payloads with unknown length or length larger than TCP_WND we need to control AsyncTCP's queue and in-flight fragmentation.
@@ -378,9 +386,6 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
   }
 #endif
 
-  // this is not functionally needed in AsyncAbstractResponse itself, but kept for compatibility if some of the derived classes are rely on it somehow
-  _ackedLength += len;
-
   size_t payloadlen{0};  // amount of data to be written to tcp sockbuff during this call, used as return value of this method
 
   // send http headers first
@@ -392,11 +397,14 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
     if (_writtenHeadersLength < _assembled_headers.length()) {
 // we were not able to fit all headers in current buff, send this part here and return later for the rest
 #if ASYNCWEBSERVER_USE_CHUNK_INFLIGHT
-      _in_flight += pcb_written;
-      --_in_flight_credit;  // take a credit
+      if (pcb_written) {
+        _in_flight += pcb_written;
+        --_in_flight_credit;  // take a credit
+      }
 #endif
-      if (!request->client()->send()) {
+      if (pcb_written && !request->client()->send()) {
         // something is wrong, what should we do here?
+        _state = RESPONSE_FAILED;
         request->client()->close();
         return 0;
       }
@@ -415,6 +423,8 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
         // data is pending in buffer from a previous call or previous iteration
         size_t const added_len =
           request->client()->add(reinterpret_cast<char *>(_send_buffer->data() + _send_buffer_offset), _send_buffer_len - _send_buffer_offset);
+        payloadlen += added_len;
+        _writtenLength += added_len;
         if (added_len != _send_buffer_len - _send_buffer_offset) {
           // we were not able to add entire buffer's content to tcp buffs, leave it for later
           // (this should not happen normally unless connection's TCP window suddenly changed from remote or mem pressure)
@@ -423,11 +433,10 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
         } else {
           _send_buffer_len = _send_buffer_offset = 0;  // consider buffer empty
         }
-        payloadlen += added_len;
       }
 
       auto tcp_win = request->client()->space();
-      if (tcp_win == 0 || _state == RESPONSE_END) {
+      if (tcp_win == 0 || _source_ended) {
         break;  // no room left or no more data
       }
 
@@ -467,7 +476,7 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
           _sentLength += readLen;           // data is not sent yet, but we won't get a chance to count this later properly for chunked data
           if (!readLen) {
             // last chunk?
-            _state = RESPONSE_END;
+            _source_ended = true;
           }
         }
       } else {
@@ -483,21 +492,24 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
 
         if (readLen == 0) {
           // no more data to send
-          _state = RESPONSE_END;
+          _source_ended = true;
         } else if (readLen != RESPONSE_TRY_AGAIN) {
           _send_buffer_len += readLen;  // set buffers's size to match added data
           _sentLength += readLen;       // data is not sent yet, but we need it to understand that it would be last block
           if (_sendContentLength && (_sentLength == _contentLength)) {
             // it was last piece of content
-            _state = RESPONSE_END;
+            _source_ended = true;
           }
         }
       }
     } while (_send_buffer_len);  // go on till we have something in buffer pending to send
 
     // execute sending whatever we have in sock buffs now
-    request->client()->send();
-    _writtenLength += payloadlen;
+    if (payloadlen && !request->client()->send()) {
+      _state = RESPONSE_FAILED;
+      request->client()->close();
+      return payloadlen;
+    }
 #if ASYNCWEBSERVER_USE_CHUNK_INFLIGHT
     // Take a credit only for data actually put in flight.  When the source
     // had nothing yet (RESPONSE_TRY_AGAIN), nothing will be acked to return
@@ -511,16 +523,13 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s
     if (_send_buffer_len == 0) {
       // buffer empty, we can release mem, otherwise need to keep it till next run (should not happen under normal conditions)
       _send_buffer.reset();
+      if (_source_ended) {
+        _state = _ackedLength >= _writtenLength ? RESPONSE_END : RESPONSE_WAIT_ACK;
+      }
     }
     return payloadlen;
   }  // (_state == RESPONSE_CONTENT)
 
-  // implicit check
-  if (_state == RESPONSE_WAIT_ACK) {
-    // we do not need to wait for any acks actually if we won't send any more data,
-    // connection would be closed gracefully with last piece of data (in AsyncWebServerRequest::_onAck)
-    _state = RESPONSE_END;
-  }
   return 0;
 }
 

tests/response_backpressure.cpp

@@ -0,0 +1,126 @@
+// SPDX-License-Identifier: LGPL-3.0-or-later
+// Host regression harness. The runner inserts the ACTUAL upstream/patched
+// AsyncAbstractResponse::write_send_buffs below, not a rewritten response model.
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <cstring>
+#include <deque>
+#include <iostream>
+#include <memory>
+#include <new>
+#include <stdexcept>
+#include <string>
+using String=std::string;
+constexpr size_t ASYNC_RESPONCE_BUFF_SIZE=16,CONFIG_LWIP_TCP_WND_DEFAULT=64,CONFIG_LWIP_TCP_MSS=32;
+constexpr size_t RESPONSE_TRY_AGAIN=static_cast<size_t>(-1);
+enum {RESPONSE_SETUP,RESPONSE_HEADERS,RESPONSE_CONTENT,RESPONSE_WAIT_ACK,RESPONSE_END,RESPONSE_FAILED};
+void check(bool value,const char *message){if(!value)throw std::runtime_error(message);}
+struct Client {
+  size_t room=64,sends=0;bool alive=true;
+  String wire,pending;std::deque<size_t> limits;std::deque<bool> sendResults;
+  size_t space(){return room;}
+  size_t add(const char *data,size_t count){
+    if(!alive)return 0;
+    size_t accepted=std::min(count,room);
+    if(!limits.empty()){accepted=std::min(accepted,limits.front());limits.pop_front();}
+    pending.append(data,accepted);room-=accepted;return accepted;
+  }
+  bool send(){
+    ++sends;bool sent=true;
+    if(!sendResults.empty()){sent=sendResults.front();sendResults.pop_front();}
+    if(sent){wire+=pending;pending.clear();}return sent;
+  }
+  void close(){alive=false;}
+};
+struct AsyncWebServerRequest {Client transport;Client *client(){return &transport;}};
+class AsyncAbstractResponse {
+public:
+  String _assembled_headers="HEAD\n",source;
+  size_t _writtenHeadersLength=0,_contentLength=0,_sentLength=0,_writtenLength=0,_ackedLength=0;
+  size_t _send_buffer_offset=0,_send_buffer_len=0,_in_flight=0,_in_flight_credit=2;
+  bool _source_ended=false,_chunked=false,_sendContentLength=true,valid=true;
+  int _state=RESPONSE_HEADERS;
+  size_t position=0,fills=0,notReady=0;
+  std::unique_ptr<std::array<uint8_t,ASYNC_RESPONCE_BUFF_SIZE>> _send_buffer;
+  bool _sourceValid()const{return valid;}
+  size_t _fillBufferAndProcessTemplates(uint8_t *data,size_t count){
+    ++fills;if(notReady){--notReady;return RESPONSE_TRY_AGAIN;}
+    const auto n=std::min(count,source.size()-position);
+    std::memcpy(data,source.data()+position,n);position+=n;return n;
+  }
+  size_t write_send_buffs(AsyncWebServerRequest*,size_t,uint32_t);
+};
+// ACTUAL_IMPLEMENTATION
+
+String unchunk(const String &wire){
+  size_t at=5;String body;
+  for(;;){check(at+6<=wire.size(),"truncated chunk header");
+    check(wire.substr(at+4,2)=="\r\n","chunk header framing");
+    const auto count=std::stoul(wire.substr(at,4),nullptr,16);at+=6;
+    check(at+count+2<=wire.size(),"truncated chunk body");body+=wire.substr(at,count);at+=count;
+    check(wire.substr(at,2)=="\r\n","chunk tail framing");at+=2;
+    if(!count){check(at==wire.size(),"duplicate chunk terminator or trailing bytes");return body;}
+  }
+}
+void finish(AsyncAbstractResponse &r,AsyncWebServerRequest &q){
+  for(unsigned budget=0;budget<200000;++budget){
+    check(q.transport.alive,"response closed while healthy");
+    check(r._state!=RESPONSE_FAILED,"healthy response failed");
+    if(r._state==RESPONSE_END){
+      check(q.transport.pending.empty(),"finished with unsent TCP data");
+      check(r._ackedLength>=q.transport.wire.size(),"finished before actual admitted bytes ACK");return;
+    }
+    const auto outstanding=q.transport.wire.size()-r._ackedLength;
+    // Real ACK callbacks can be smaller than the last admitted tail.
+    const auto ack=std::min<size_t>(outstanding,3);q.transport.room=64;
+    r.write_send_buffs(&q,ack,0);
+  }
+  throw std::runtime_error("response failed to progress under ACK/poll");
+}
+void content(const String &body,bool chunked,bool known){
+  std::cout<<"case bytes="<<body.size()<<" chunked="<<chunked<<" known="<<known<<'\n';
+  AsyncWebServerRequest q;AsyncAbstractResponse r;
+  r.source=body;r._contentLength=known?body.size():0;r._chunked=chunked;r._sendContentLength=known;
+  // Both final-buffer short writes and zero-admission polls occur while the
+  // reported space remains larger than the accepted write (memory pressure).
+  q.transport.limits={5,1,0,2,1,0,4};r.write_send_buffs(&q,0,0);
+  check(r._state!=RESPONSE_END,"source EOF was mistaken for drained TCP output");
+  finish(r,q);
+  check(r._writtenLength==q.transport.wire.size(),"TCP accounting misses partial add or double-counts headers");
+  check(q.transport.wire.substr(0,5)=="HEAD\n","header changed");
+  if(!(chunked?unchunk(q.transport.wire)==body:q.transport.wire=="HEAD\n"+body))
+    throw std::runtime_error("byte-exact body lost or duplicated: wire="+std::to_string(q.transport.wire.size())+" wanted="+std::to_string(body.size()+5));
+}
+int main(){try{
+  // Baseline fails here deterministically: source filled8, TCP admitted1,
+  // RESPONSE_END is exposed while7 bytes remain in the local send buffer.
+  content("12345678",false,true);
+  content("",false,true);
+  content(String(97,'x'),false,true);
+  content(String(224*1024,'p'),false,true);
+  content("unknown length body",false,false);
+  content("",true,false);content(String(97,'c'),true,false);
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;r.source="body";r._contentLength=4;
+    q.transport.limits={0,0,5,0,4};r.write_send_buffs(&q,0,0);
+    check(q.transport.sends==0&&q.transport.alive,"zero header add tried send or closed");
+    finish(r,q);check(q.transport.wire=="HEAD\nbody","zero header retry changed bytes"); }
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;r.source="window";r._contentLength=6;q.transport.room=0;
+    r.write_send_buffs(&q,0,0);check(q.transport.alive&&q.transport.wire.empty()&&q.transport.sends==0,"zero TCP window closed or sent");
+    finish(r,q);check(q.transport.wire=="HEAD\nwindow","zero TCP window lost body"); }
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;r.source="ready later";r._sendContentLength=false;r.notReady=3;
+    r.write_send_buffs(&q,0,0);finish(r,q);check(q.transport.wire=="HEAD\nready later","TRY_AGAIN exhausted credits"); }
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;r.source="1234";r._contentLength=4;
+    r.write_send_buffs(&q,0,0);check(r._state==RESPONSE_WAIT_ACK,"did not wait for final ACK");
+    const auto admitted=q.transport.wire.size();r.write_send_buffs(&q,admitted-1,0);
+    check(r._state==RESPONSE_WAIT_ACK,"early partial ACK completed response");
+    r.write_send_buffs(&q,0,0);check(r._state==RESPONSE_WAIT_ACK,"poll completed unacknowledged tail");
+    r.write_send_buffs(&q,1,0);check(r._state==RESPONSE_END,"final actual ACK did not complete"); }
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;r.source="1234";r._contentLength=4;q.transport.sendResults={false};
+    r.write_send_buffs(&q,0,0);check(r._state==RESPONSE_FAILED&&!q.transport.alive,"failed send masqueraded as success"); }
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;q.transport.limits={1};q.transport.sendResults={false};
+    r.write_send_buffs(&q,0,0);check(r._state==RESPONSE_FAILED&&!q.transport.alive,"partial header send failure not explicit"); }
+  { AsyncWebServerRequest q;AsyncAbstractResponse r;r.valid=false;
+    r.write_send_buffs(&q,0,0);check(r._state==RESPONSE_FAILED&&!q.transport.alive&&q.transport.wire.empty(),"invalid source was transmitted"); }
+  std::cout<<"Actual extracted response: partial/zero header+body, final ACK, empty/224KiB/unknown/chunked, TRY_AGAIN, send failure and invalid source pass\n";return 0;
+}catch(const std::exception &e){std::cerr<<"FAIL: "<<e.what()<<'\n';return 1;}}

tests/run-response-backpressure.mjs

@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: LGPL-3.0-or-later
+// Run: node tests/run-response-backpressure.mjs [--baseline WebResponses.cpp] [--output EXECUTABLE]
+// No SDK/library installation, download, board contact or extra compiler cache.
+// Compile the actual method from the exact reviewed source with transport mocks.
+import {readFileSync,mkdirSync} from 'node:fs';
+import {resolve,dirname} from 'node:path';
+import {fileURLToPath} from 'node:url';
+import {spawnSync} from 'node:child_process';
+import {createHash} from 'node:crypto';
+const here=dirname(fileURLToPath(import.meta.url)),root=resolve(here,'..');
+const options={};for(let i=2;i<process.argv.length;i+=2){
+  if(!['--baseline','--output'].includes(process.argv[i])||!process.argv[i+1])throw Error('Use optional --baseline SOURCE and --output EXECUTABLE');
+  options[process.argv[i]]=resolve(process.argv[i+1]);
+}
+const out=options['--output']||resolve(root,'build/validation/response-backpressure'+(process.platform==='win32'?'.exe':''));
+mkdirSync(dirname(out),{recursive:true});
+const harness=readFileSync(resolve(here,'response_backpressure.cpp'),'utf8');
+if(!readFileSync(resolve(root,'src/WebResponseImpl.h'),'utf8').includes('bool _source_ended{false};'))throw Error('Patched production response is missing its EOF state');
+function method(path,baseline){
+  const source=readFileSync(path,'utf8'),start=source.indexOf('size_t AsyncAbstractResponse::write_send_buffs('),
+      end=source.indexOf('size_t AsyncAbstractResponse::_readDataFromCacheOrContent(',start);
+  if(start<0||end<0)throw Error('Actual method boundaries changed; no fixture executed');
+  const digest=createHash('sha256').update(source).digest('hex');
+  if(baseline&&digest!=='ec9481a3183e53f3ec56a3ab22faae926403aadaf3afb4e41551d04f915d75e7')throw Error('Baseline is not reviewed upstream4d4eaca source');
+  console.log(path+' sha256='+digest);
+  return harness.replace('// ACTUAL_IMPLEMENTATION',()=>source.slice(start,end));
+}
+for(const baseline of options['--baseline']?[true,false]:[false])for(const credits of [0,1]){
+  const path=baseline?options['--baseline']:resolve(root,'src/WebResponses.cpp');
+  const compiled=spawnSync(process.env.CXX||'g++',['-std=c++17','-Wall','-Wextra','-Werror','-O0','-g0',
+    '-DASYNCWEBSERVER_USE_CHUNK_INFLIGHT='+credits,'-x','c++','-','-o',out],
+    {input:method(path,baseline),encoding:'utf8',windowsHide:true,timeout:30000});
+  if(compiled.error||compiled.status!==0)throw Error('Fixture compile failed: '+compiled.stderr);
+  const result=spawnSync(out,[],{encoding:'utf8',windowsHide:true,timeout:20000});
+  console.log((baseline?'BASELINE':'PATCHED')+' credits='+credits+' exit='+result.status+' '+result.stdout+result.stderr);
+  if(result.error||result.status!==(baseline?1:0))throw Error('Unexpected baseline/patch fixture outcome');
+}