buffer: add "overflow" watermark

include/envoy/buffer/buffer.h

@@ -372,10 +372,13 @@ class WatermarkFactory {
    *   low watermark.
    * @param above_high_watermark supplies a function to call if the buffer goes over a configured
    *   high watermark.
+   * @param above_overflow_watermark supplies a function to call if the buffer goes over a
+   *   configured "overflow" watermark.
    * @return a newly created InstancePtr.
    */
   virtual InstancePtr create(std::function<void()> below_low_watermark,
-                             std::function<void()> above_high_watermark) PURE;
+                             std::function<void()> above_high_watermark,
+                             std::function<void()> above_overflow_watermark) PURE;
 };
 
 using WatermarkFactoryPtr = std::unique_ptr<WatermarkFactory>;

include/envoy/http/codec.h

@@ -139,6 +139,12 @@ class StreamCallbacks {
   virtual void onResetStream(StreamResetReason reason,
                              absl::string_view transport_failure_reason) PURE;
 
+  /**
+   * Fires when a stream, or the connection the stream is sending to, goes over its "overflow"
+   * watermark.
+   */
+  virtual void onAboveWriteBufferOverflowWatermark() PURE;
+
   /**
    * Fires when a stream, or the connection the stream is sending to, goes over its high watermark.
    */
@@ -310,6 +316,11 @@ class Connection {
    */
   virtual bool wantsToWrite() PURE;
 
+  /**
+   * Called when the underlying Network::Connection goes over its "overflow" watermark.
+   */
+  virtual void onUnderlyingConnectionAboveWriteBufferOverflowWatermark() PURE;
+
   /**
    * Called when the underlying Network::Connection goes over its high watermark.
    */
@@ -329,6 +340,13 @@ class DownstreamWatermarkCallbacks {
 public:
   virtual ~DownstreamWatermarkCallbacks() = default;
 
+  /**
+   * Called when the downstream connection or stream goes over its "overflow" watermark. Note that
+   * this may be called separately for both the stream going over and the connection going over.
+   * The implementation should close the stream.
+   */
+  virtual void onAboveWriteBufferOverflowWatermark() PURE;
+
   /**
    * Called when the downstream connection or stream goes over its high watermark. Note that this
    * may be called separately for both the stream going over and the connection going over. It

include/envoy/http/filter.h

@@ -354,7 +354,14 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
   virtual void encodeMetadata(MetadataMapPtr&& metadata_map) PURE;
 
   /**
-   * Called when the buffer for a decoder filter or any buffers the filter sends data to go over
+   * Called when the buffer for a decoder filter, or any buffers the filter sends data to, go over
+   * their "overflow" watermark. Implementations should close/reset any streams that overflow their
+   * write buffers.
+   */
+  virtual void onDecoderFilterAboveWriteBufferOverflowWatermark() PURE;
+
+  /**
+   * Called when the buffer for a decoder filter, or any buffers the filter sends data to, go over
    * their high watermark.
    *
    * In the case of a filter such as the router filter, which spills into multiple buffers (codec,
@@ -603,6 +610,12 @@ class StreamEncoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual HeaderMap& addEncodedTrailers() PURE;
 
+  /**
+   * Called when an encoder filter goes over its "overflow" watermark. The stream should be closed
+   * in response to overflows.
+   */
+  virtual void onEncoderFilterAboveWriteBufferOverflowWatermark() PURE;
+
   /**
    * Called when an encoder filter goes over its high watermark.
    */

include/envoy/network/connection.h

@@ -47,6 +47,12 @@ class ConnectionCallbacks {
    */
   virtual void onEvent(ConnectionEvent event) PURE;
 
+  /**
+   * Called when the write buffer for a connection goes over its "overflow"
+   * watermark.
+   */
+  virtual void onAboveWriteBufferOverflowWatermark() PURE;
+
   /**
    * Called when the write buffer for a connection goes over its high watermark.
    */

source/common/buffer/watermark_buffer.cc

@@ -7,32 +7,32 @@ namespace Buffer {
 
 void WatermarkBuffer::add(const void* data, uint64_t size) {
   OwnedImpl::add(data, size);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::add(absl::string_view data) {
   OwnedImpl::add(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::add(const Instance& data) {
   OwnedImpl::add(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::prepend(absl::string_view data) {
   OwnedImpl::prepend(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::prepend(Instance& data) {
   OwnedImpl::prepend(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::commit(RawSlice* iovecs, uint64_t num_iovecs) {
   OwnedImpl::commit(iovecs, num_iovecs);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::drain(uint64_t size) {
@@ -42,23 +42,23 @@ void WatermarkBuffer::drain(uint64_t size) {
 
 void WatermarkBuffer::move(Instance& rhs) {
   OwnedImpl::move(rhs);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::move(Instance& rhs, uint64_t length) {
   OwnedImpl::move(rhs, length);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 Api::IoCallUint64Result WatermarkBuffer::read(Network::IoHandle& io_handle, uint64_t max_length) {
   Api::IoCallUint64Result result = OwnedImpl::read(io_handle, max_length);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
   return result;
 }
 
 uint64_t WatermarkBuffer::reserve(uint64_t length, RawSlice* iovecs, uint64_t num_iovecs) {
   uint64_t bytes_reserved = OwnedImpl::reserve(length, iovecs, num_iovecs);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
   return bytes_reserved;
 }
 
@@ -68,11 +68,14 @@ Api::IoCallUint64Result WatermarkBuffer::write(Network::IoHandle& io_handle) {
   return result;
 }
 
-void WatermarkBuffer::setWatermarks(uint32_t low_watermark, uint32_t high_watermark) {
-  ASSERT(low_watermark < high_watermark || (high_watermark == 0 && low_watermark == 0));
+void WatermarkBuffer::setWatermarks(uint32_t low_watermark, uint32_t high_watermark,
+                                    uint32_t overflow_watermark) {
+  ASSERT((low_watermark < high_watermark && high_watermark < overflow_watermark) ||
+         (overflow_watermark == 0 && high_watermark == 0 && low_watermark == 0));
   low_watermark_ = low_watermark;
   high_watermark_ = high_watermark;
-  checkHighWatermark();
+  overflow_watermark_ = overflow_watermark;
+  checkHighAndOverflowWatermarks();
   checkLowWatermark();
 }
 
@@ -86,14 +89,19 @@ void WatermarkBuffer::checkLowWatermark() {
   below_low_watermark_();
 }
 
-void WatermarkBuffer::checkHighWatermark() {
-  if (above_high_watermark_called_ || high_watermark_ == 0 ||
-      OwnedImpl::length() <= high_watermark_) {
+void WatermarkBuffer::checkHighAndOverflowWatermarks() {
+  if (!above_overflow_watermark_called_ && overflow_watermark_ != 0 &&
+      OwnedImpl::length() > overflow_watermark_) {
+    above_overflow_watermark_called_ = true;
+    above_overflow_watermark_();
     return;
   }
 
-  above_high_watermark_called_ = true;
-  above_high_watermark_();
+  if (!above_high_watermark_called_ && high_watermark_ != 0 &&
+      OwnedImpl::length() > high_watermark_) {
+    above_high_watermark_called_ = true;
+    above_high_watermark_();
+  }
 }
 
 } // namespace Buffer

source/common/buffer/watermark_buffer.h

@@ -16,8 +16,10 @@ namespace Buffer {
 class WatermarkBuffer : public OwnedImpl {
 public:
   WatermarkBuffer(std::function<void()> below_low_watermark,
-                  std::function<void()> above_high_watermark)
-      : below_low_watermark_(below_low_watermark), above_high_watermark_(above_high_watermark) {}
+                  std::function<void()> above_high_watermark,
+                  std::function<void()> above_overflow_watermark)
+      : below_low_watermark_(below_low_watermark), above_high_watermark_(above_high_watermark),
+        above_overflow_watermark_(above_overflow_watermark) {}
 
   // Override all functions from Instance which can result in changing the size
   // of the underlying buffer.
@@ -35,25 +37,29 @@ class WatermarkBuffer : public OwnedImpl {
   Api::IoCallUint64Result write(Network::IoHandle& io_handle) override;
   void postProcess() override { checkLowWatermark(); }
 
-  void setWatermarks(uint32_t watermark) { setWatermarks(watermark / 2, watermark); }
-  void setWatermarks(uint32_t low_watermark, uint32_t high_watermark);
+  void setWatermarks(uint32_t watermark) { setWatermarks(watermark / 2, watermark, watermark * 2); }
+  void setWatermarks(uint32_t low_watermark, uint32_t high_watermark, uint32_t overflow_watermark);
   uint32_t highWatermark() const { return high_watermark_; }
 
 private:
-  void checkHighWatermark();
+  void checkHighAndOverflowWatermarks();
   void checkLowWatermark();
 
   std::function<void()> below_low_watermark_;
   std::function<void()> above_high_watermark_;
+  std::function<void()> above_overflow_watermark_;
 
   // Used for enforcing buffer limits (off by default). If these are set to non-zero by a call to
   // setWatermarks() the watermark callbacks will be called as described above.
+  uint32_t overflow_watermark_{0};
   uint32_t high_watermark_{0};
   uint32_t low_watermark_{0};
-  // Tracks the latest state of watermark callbacks.
-  // True between the time above_high_watermark_ has been called until above_high_watermark_ has
-  // been called.
+  // Set to true after above_high_watermark_ has been called, and reset to false after
+  // below_low_watermark_ has been called.
   bool above_high_watermark_called_{false};
+  // Set to true after above_overflow_watermark_ has been called. Never reset, because we assume
+  // the stream will be forcibly closed in response.
+  bool above_overflow_watermark_called_{false};
 };
 
 using WatermarkBufferPtr = std::unique_ptr<WatermarkBuffer>;
@@ -62,8 +68,10 @@ class WatermarkBufferFactory : public WatermarkFactory {
 public:
   // Buffer::WatermarkFactory
   InstancePtr create(std::function<void()> below_low_watermark,
-                     std::function<void()> above_high_watermark) override {
-    return InstancePtr{new WatermarkBuffer(below_low_watermark, above_high_watermark)};
+                     std::function<void()> above_high_watermark,
+                     std::function<void()> above_overflow_watermark) override {
+    return InstancePtr{
+        new WatermarkBuffer(below_low_watermark, above_high_watermark, above_overflow_watermark)};
   }
 };
 

source/common/http/async_client_impl.h

@@ -333,6 +333,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   void encodeData(Buffer::Instance& data, bool end_stream) override;
   void encodeTrailers(HeaderMapPtr&& trailers) override;
   void encodeMetadata(MetadataMapPtr&&) override {}
+  void onDecoderFilterAboveWriteBufferOverflowWatermark() override {}
   void onDecoderFilterAboveWriteBufferHighWatermark() override {}
   void onDecoderFilterBelowWriteBufferLowWatermark() override {}
   void addDownstreamWatermarkCallbacks(DownstreamWatermarkCallbacks&) override {}

source/common/http/codec_client.h

@@ -192,6 +192,7 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
     void onResetStream(StreamResetReason reason, absl::string_view) override {
       parent_.onReset(*this, reason);
     }
+    void onAboveWriteBufferOverflowWatermark() override {}
     void onAboveWriteBufferHighWatermark() override {}
     void onBelowWriteBufferLowWatermark() override {}
 
@@ -219,6 +220,9 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
   void onEvent(Network::ConnectionEvent event) override;
   // Pass watermark events from the connection on to the codec which will pass it to the underlying
   // streams.
+  void onAboveWriteBufferOverflowWatermark() override {
+    codec_->onUnderlyingConnectionAboveWriteBufferOverflowWatermark();
+  }
   void onAboveWriteBufferHighWatermark() override {
     codec_->onUnderlyingConnectionAboveWriteBufferHighWatermark();
   }

source/common/http/codec_helper.h

@@ -36,6 +36,17 @@ class StreamCallbackHelper {
     }
   }
 
+  void runOverflowWatermarkCallbacks() {
+    if (reset_callbacks_started_ || local_end_stream_) {
+      return;
+    }
+    for (StreamCallbacks* callbacks : callbacks_) {
+      if (callbacks) {
+        callbacks->onAboveWriteBufferOverflowWatermark();
+      }
+    }
+  }
+
   void runResetCallbacks(StreamResetReason reason) {
     // Reset callbacks are a special case, and the only StreamCallbacks allowed
     // to run after local_end_stream_.

source/common/http/conn_manager_impl.cc

@@ -1649,6 +1649,11 @@ void ConnectionManagerImpl::ActiveStream::onResetStream(StreamResetReason, absl:
   connection_manager_.doDeferredStreamDestroy(*this);
 }
 
+void ConnectionManagerImpl::ActiveStream::onAboveWriteBufferOverflowWatermark() {
+  ENVOY_STREAM_LOG(debug, "Closing upstream stream due to downstream stream overflow.", *this);
+  callOverflowWatermarkCallbacks();
+}
+
 void ConnectionManagerImpl::ActiveStream::onAboveWriteBufferHighWatermark() {
   ENVOY_STREAM_LOG(debug, "Disabling upstream stream due to downstream stream watermark.", *this);
   callHighWatermarkCallbacks();
@@ -1672,6 +1677,12 @@ bool ConnectionManagerImpl::ActiveStream::verbose() const {
   return connection_manager_.config_.tracingConfig()->verbose_;
 }
 
+void ConnectionManagerImpl::ActiveStream::callOverflowWatermarkCallbacks() {
+  for (auto watermark_callbacks : watermark_callbacks_) {
+    watermark_callbacks->onAboveWriteBufferOverflowWatermark();
+  }
+}
+
 void ConnectionManagerImpl::ActiveStream::callHighWatermarkCallbacks() {
   ++high_watermark_count_;
   for (auto watermark_callbacks : watermark_callbacks_) {
@@ -1939,9 +1950,9 @@ void ConnectionManagerImpl::ActiveStreamFilterBase::clearRouteCache() {
 }
 
 Buffer::WatermarkBufferPtr ConnectionManagerImpl::ActiveStreamDecoderFilter::createBuffer() {
-  auto buffer =
-      std::make_unique<Buffer::WatermarkBuffer>([this]() -> void { this->requestDataDrained(); },
-                                                [this]() -> void { this->requestDataTooLarge(); });
+  auto buffer = std::make_unique<Buffer::WatermarkBuffer>(
+      [this]() -> void { this->requestDataDrained(); },
+      [this]() -> void { this->requestDataTooLarge(); }, [this]() -> void { this->resetStream(); });
   buffer->setWatermarks(parent_.buffer_limit_);
   return buffer;
 }
@@ -2013,6 +2024,13 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::encodeMetadata(
   parent_.encodeMetadata(nullptr, std::move(metadata_map_ptr));
 }
 
+void ConnectionManagerImpl::ActiveStreamDecoderFilter::
+    onDecoderFilterAboveWriteBufferOverflowWatermark() {
+  ENVOY_STREAM_LOG(debug, "Closing downstream stream due to filter callbacks.", parent_);
+  // TODO(mergeconflict): Add a new flow control stat.
+  resetStream();
+}
+
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::
     onDecoderFilterAboveWriteBufferHighWatermark() {
   ENVOY_STREAM_LOG(debug, "Read-disabling downstream stream due to filter callbacks.", parent_);
@@ -2086,7 +2104,8 @@ bool ConnectionManagerImpl::ActiveStreamDecoderFilter::recreateStream() {
 
 Buffer::WatermarkBufferPtr ConnectionManagerImpl::ActiveStreamEncoderFilter::createBuffer() {
   auto buffer = new Buffer::WatermarkBuffer([this]() -> void { this->responseDataDrained(); },
-                                            [this]() -> void { this->responseDataTooLarge(); });
+                                            [this]() -> void { this->responseDataTooLarge(); },
+                                            [this]() -> void { this->resetStream(); });
   buffer->setWatermarks(parent_.buffer_limit_);
   return Buffer::WatermarkBufferPtr{buffer};
 }
@@ -2106,6 +2125,12 @@ HeaderMap& ConnectionManagerImpl::ActiveStreamEncoderFilter::addEncodedTrailers(
   return parent_.addEncodedTrailers();
 }
 
+void ConnectionManagerImpl::ActiveStreamEncoderFilter::
+    onEncoderFilterAboveWriteBufferOverflowWatermark() {
+  ENVOY_STREAM_LOG(debug, "Closing upstream stream due to filter callbacks.", parent_);
+  parent_.callOverflowWatermarkCallbacks();
+}
+
 void ConnectionManagerImpl::ActiveStreamEncoderFilter::
     onEncoderFilterAboveWriteBufferHighWatermark() {
   ENVOY_STREAM_LOG(debug, "Disabling upstream stream due to filter callbacks.", parent_);