wasm: fix fail-close streams on VM failure.

bazel/repository_locations.bzl

@@ -948,8 +948,8 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         project_name = "WebAssembly for Proxies (C++ host implementation)",
         project_desc = "WebAssembly for Proxies (C++ host implementation)",
         project_url = "https://github.com/proxy-wasm/proxy-wasm-cpp-host",
-        version = "4b33df7638637fcc680291daa9d7a57c59e0411e",
-        sha256 = "e78507e04dc8b154ced5b19d6175bd1435fd850fdd127bce541d31799db06cd4",
+        version = "579189940ee48ebf1fb1d6539483506bee89f0b4",
+        sha256 = "eedcc5d0e73a715d9361eda39b21b178e077ab4d749d6bf2f030de81f668d6d3",
         strip_prefix = "proxy-wasm-cpp-host-{version}",
         urls = ["https://github.com/proxy-wasm/proxy-wasm-cpp-host/archive/{version}.tar.gz"],
         use_category = ["dataplane_ext"],
@@ -964,7 +964,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
             "envoy.wasm.runtime.wavm",
             "envoy.wasm.runtime.wasmtime",
         ],
-        release_date = "2021-04-20",
+        release_date = "2021-04-28",
         cpe = "N/A",
     ),
     proxy_wasm_rust_sdk = dict(

source/extensions/common/wasm/context.cc

@@ -88,8 +88,6 @@ Http::RequestHeaderMapPtr buildRequestHeaderMapFromPairs(const Pairs& pairs) {
 
 template <typename P> static uint32_t headerSize(const P& p) { return p ? p->size() : 0; }
 
-constexpr absl::string_view FailStreamResponseDetails = "wasm_fail_stream";
-
 } // namespace
 
 // Test support.
@@ -1584,20 +1582,22 @@ WasmResult Context::continueStream(WasmStreamType stream_type) {
   return WasmResult::Ok;
 }
 
+constexpr absl::string_view CloseStreamResponseDetails = "wasm_close_stream";
+
 WasmResult Context::closeStream(WasmStreamType stream_type) {
   switch (stream_type) {
   case WasmStreamType::Request:
     if (decoder_callbacks_) {
       if (!decoder_callbacks_->streamInfo().responseCodeDetails().has_value()) {
-        decoder_callbacks_->streamInfo().setResponseCodeDetails(FailStreamResponseDetails);
+        decoder_callbacks_->streamInfo().setResponseCodeDetails(CloseStreamResponseDetails);
       }
       decoder_callbacks_->resetStream();
     }
     return WasmResult::Ok;
   case WasmStreamType::Response:
     if (encoder_callbacks_) {
       if (!encoder_callbacks_->streamInfo().responseCodeDetails().has_value()) {
-        encoder_callbacks_->streamInfo().setResponseCodeDetails(FailStreamResponseDetails);
+        encoder_callbacks_->streamInfo().setResponseCodeDetails(CloseStreamResponseDetails);
       }
       encoder_callbacks_->resetStream();
     }
@@ -1618,6 +1618,39 @@ WasmResult Context::closeStream(WasmStreamType stream_type) {
   return WasmResult::BadArgument;
 }
 
+constexpr absl::string_view FailStreamResponseDetails = "wasm_fail_stream";
+
+void Context::failStream(WasmStreamType stream_type) {
+  switch (stream_type) {
+  case WasmStreamType::Request:
+    if (decoder_callbacks_) {
+      decoder_callbacks_->sendLocalReply(Envoy::Http::Code::ServiceUnavailable, "", nullptr,
+                                         Grpc::Status::WellKnownGrpcStatus::Unavailable,
+                                         FailStreamResponseDetails);
+    }
+    break;
+  case WasmStreamType::Response:
+    if (encoder_callbacks_) {
+      encoder_callbacks_->sendLocalReply(Envoy::Http::Code::ServiceUnavailable, "", nullptr,
+                                         Grpc::Status::WellKnownGrpcStatus::Unavailable,
+                                         FailStreamResponseDetails);
+    }
+    break;
+  case WasmStreamType::Downstream:
+    if (network_read_filter_callbacks_) {
+      network_read_filter_callbacks_->connection().close(
+          Envoy::Network::ConnectionCloseType::FlushWrite);
+    }
+    break;
+  case WasmStreamType::Upstream:
+    if (network_write_filter_callbacks_) {
+      network_write_filter_callbacks_->connection().close(
+          Envoy::Network::ConnectionCloseType::FlushWrite);
+    }
+    break;
+  }
+}
+
 WasmResult Context::sendLocalResponse(uint32_t response_code, absl::string_view body_text,
                                       Pairs additional_headers, uint32_t grpc_status,
                                       absl::string_view details) {

source/extensions/common/wasm/context.h

@@ -209,6 +209,7 @@ class Context : public proxy_wasm::ContextBase,
   // Continue
   WasmResult continueStream(WasmStreamType stream_type) override;
   WasmResult closeStream(WasmStreamType stream_type) override;
+  void failStream(WasmStreamType stream_type) override;
   WasmResult sendLocalResponse(uint32_t response_code, absl::string_view body_text,
                                Pairs additional_headers, uint32_t grpc_status,
                                absl::string_view details) override;

test/extensions/filters/http/wasm/BUILD

@@ -27,6 +27,7 @@ envoy_extension_cc_test(
         "//test/extensions/filters/http/wasm/test_data:body_rust.wasm",
         "//test/extensions/filters/http/wasm/test_data:headers_rust.wasm",
         "//test/extensions/filters/http/wasm/test_data:metadata_rust.wasm",
+        "//test/extensions/filters/http/wasm/test_data:panic_rust.wasm",
         "//test/extensions/filters/http/wasm/test_data:resume_call_rust.wasm",
         "//test/extensions/filters/http/wasm/test_data:shared_data_rust.wasm",
         "//test/extensions/filters/http/wasm/test_data:shared_queue_rust.wasm",

test/extensions/filters/http/wasm/config_test.cc

@@ -837,9 +837,12 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteSuccessBadcode) {
 
   context->setDecoderFilterCallbacks(decoder_callbacks);
   EXPECT_CALL(decoder_callbacks, streamInfo()).WillRepeatedly(ReturnRef(stream_info));
-  EXPECT_CALL(stream_info, setResponseCodeDetails("wasm_fail_stream"));
-  EXPECT_CALL(decoder_callbacks, resetStream());
-
+  auto headers = Http::TestResponseHeaderMapImpl{{":status", "503"}};
+  EXPECT_CALL(decoder_callbacks, encodeHeaders_(HeaderMapEqualRef(&headers), true));
+  EXPECT_CALL(decoder_callbacks,
+              sendLocalReply(Envoy::Http::Code::ServiceUnavailable, testing::Eq(""), _,
+                             testing::Eq(Grpc::Status::WellKnownGrpcStatus::Unavailable),
+                             testing::Eq("wasm_fail_stream")));
   EXPECT_EQ(context->onRequestHeaders(10, false),
             proxy_wasm::FilterHeadersStatus::StopAllIterationAndWatermark);
 }

test/extensions/filters/http/wasm/test_data/BUILD

@@ -48,6 +48,15 @@ wasm_rust_binary(
     ],
 )
 
+wasm_rust_binary(
+    name = "panic_rust.wasm",
+    srcs = ["panic_rust.rs"],
+    deps = [
+        "@proxy_wasm_rust_sdk//:proxy_wasm",
+        "@proxy_wasm_rust_sdk//bazel/cargo:log",
+    ],
+)
+
 wasm_rust_binary(
     name = "resume_call_rust.wasm",
     srcs = ["resume_call_rust.rs"],
@@ -80,6 +89,7 @@ envoy_cc_library(
     srcs = [
         "test_async_call_cpp.cc",
         "test_body_cpp.cc",
+        "test_close_stream_cpp.cc",
         "test_cpp.cc",
         "test_cpp_null_plugin.cc",
         "test_grpc_call_cpp.cc",
@@ -107,9 +117,11 @@ envoy_wasm_cc_binary(
     srcs = [
         "test_async_call_cpp.cc",
         "test_body_cpp.cc",
+        "test_close_stream_cpp.cc",
         "test_cpp.cc",
         "test_grpc_call_cpp.cc",
         "test_grpc_stream_cpp.cc",
+        "test_panic_cpp.cc",
         "test_resume_call_cpp.cc",
         "test_shared_data_cpp.cc",
         "test_shared_queue_cpp.cc",

test/extensions/filters/http/wasm/test_data/panic_rust.rs

@@ -0,0 +1,39 @@
+use proxy_wasm::traits::{Context, HttpContext};
+use proxy_wasm::types::*;
+
+#[no_mangle]
+pub fn _start() {
+    proxy_wasm::set_http_context(|_, _| -> Box<dyn HttpContext> {
+        Box::new(TestStream{})
+    });
+}
+
+struct TestStream {}
+
+impl Context for TestStream {}
+
+impl HttpContext for TestStream {
+    fn on_http_request_headers(&mut self, _: usize) -> Action {
+        panic!("");
+    }
+
+    fn on_http_request_body(&mut self, _: usize, _: bool) -> Action {
+        panic!("");
+    }
+
+    fn on_http_request_trailers(&mut self, _: usize) -> Action {
+        panic!("");
+    }
+
+    fn on_http_response_headers(&mut self, _: usize) -> Action {
+        panic!("");
+    }
+
+    fn on_http_response_body(&mut self, _: usize, _: bool) -> Action {
+        panic!("");
+    }
+
+    fn on_http_response_trailers(&mut self, _: usize) -> Action {
+        panic!("");
+    }
+}

test/extensions/filters/http/wasm/test_data/test_close_stream_cpp.cc

@@ -0,0 +1,40 @@
+// NOLINT(namespace-envoy)
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#ifndef NULL_PLUGIN
+#include "proxy_wasm_intrinsics_lite.h"
+#else
+#include "extensions/common/wasm/ext/envoy_null_plugin.h"
+#endif
+
+START_WASM_PLUGIN(HttpWasmTestCpp)
+
+class CloseStreamRootContext : public RootContext {
+public:
+  explicit CloseStreamRootContext(uint32_t id, std::string_view root_id) : RootContext(id, root_id) {}
+};
+
+class CloseStreamContext : public Context {
+public:
+  explicit CloseStreamContext(uint32_t id, RootContext* root) : Context(id, root) {}
+
+  FilterHeadersStatus onRequestHeaders(uint32_t, bool) override;
+  FilterHeadersStatus onResponseHeaders(uint32_t, bool) override;
+};
+
+static RegisterContextFactory register_CloseStreamContext(CONTEXT_FACTORY(CloseStreamContext),
+                                                   ROOT_FACTORY(CloseStreamRootContext), "close_stream");
+
+FilterHeadersStatus CloseStreamContext::onRequestHeaders(uint32_t, bool) {
+  closeRequest();
+  return FilterHeadersStatus::Continue;
+}
+
+FilterHeadersStatus CloseStreamContext::onResponseHeaders(uint32_t, bool) {
+  closeResponse();
+  return FilterHeadersStatus::Continue;
+}
+
+END_WASM_PLUGIN

test/extensions/filters/http/wasm/test_data/test_panic_cpp.cc

@@ -0,0 +1,66 @@
+// NOLINT(namespace-envoy)
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#ifndef NULL_PLUGIN
+#include "proxy_wasm_intrinsics_lite.h"
+#else
+#include "extensions/common/wasm/ext/envoy_null_plugin.h"
+#endif
+
+START_WASM_PLUGIN(HttpWasmTestCpp)
+
+class PanicRootContext : public RootContext {
+public:
+  explicit PanicRootContext(uint32_t id, std::string_view root_id) : RootContext(id, root_id) {}
+};
+
+class PanicContext : public Context {
+public:
+  explicit PanicContext(uint32_t id, RootContext* root) : Context(id, root) {}
+
+  FilterHeadersStatus onRequestHeaders(uint32_t, bool) override;
+  FilterDataStatus onRequestBody(size_t , bool ) override;
+  FilterTrailersStatus onRequestTrailers(uint32_t) override;
+  FilterHeadersStatus onResponseHeaders(uint32_t, bool) override;
+  FilterDataStatus onResponseBody(size_t, bool) override;
+  FilterTrailersStatus onResponseTrailers(uint32_t) override;
+};
+
+static RegisterContextFactory register_PanicContext(CONTEXT_FACTORY(PanicContext),
+                                                   ROOT_FACTORY(PanicRootContext), "panic");
+
+static int* badptr = nullptr;
+
+FilterHeadersStatus PanicContext::onRequestHeaders(uint32_t, bool) {
+  *badptr = 0;
+  return FilterHeadersStatus::Continue;
+}
+
+FilterHeadersStatus PanicContext::onResponseHeaders(uint32_t, bool) {
+  *badptr = 0;
+  return FilterHeadersStatus::Continue;
+}
+
+FilterTrailersStatus PanicContext::onRequestTrailers(uint32_t) {
+  *badptr = 0;
+  return FilterTrailersStatus::Continue;
+}
+
+FilterDataStatus PanicContext::onRequestBody(size_t , bool ) {
+  *badptr = 0;
+  return FilterDataStatus::Continue;
+}
+
+FilterDataStatus PanicContext::onResponseBody(size_t, bool) {
+  *badptr = 0;
+  return FilterDataStatus::Continue;
+}
+
+FilterTrailersStatus PanicContext::onResponseTrailers(uint32_t) {
+  *badptr = 0;
+  return FilterTrailersStatus::Continue;
+}
+
+END_WASM_PLUGIN