src/shrpx_http2_upstream.cc

   1 /*
   2  * nghttp2 - HTTP/2 C Library
   3  *
   4  * Copyright (c) 2012 Tatsuhiro Tsujikawa
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining
   7  * a copy of this software and associated documentation files (the
   8  * "Software"), to deal in the Software without restriction, including
   9  * without limitation the rights to use, copy, modify, merge, publish,
  10  * distribute, sublicense, and/or sell copies of the Software, and to
  11  * permit persons to whom the Software is furnished to do so, subject to
  12  * the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be
  15  * included in all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  21  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  22  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  23  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  24  */
  25 #include "shrpx_http2_upstream.h"
  26
  27 #include <netinet/tcp.h>
  28 #include <assert.h>
  29 #include <cerrno>
  30 #include <sstream>
  31
  32 #include "shrpx_client_handler.h"
  33 #include "shrpx_https_upstream.h"
  34 #include "shrpx_downstream.h"
  35 #include "shrpx_downstream_connection.h"
  36 #include "shrpx_config.h"
  37 #include "shrpx_http.h"
  38 #include "shrpx_worker.h"
  39 #include "shrpx_http2_session.h"
  40 #include "shrpx_log.h"
  41 #ifdef HAVE_MRUBY
  42 #  include "shrpx_mruby.h"
  43 #endif // HAVE_MRUBY
  44 #include "http2.h"
  45 #include "util.h"
  46 #include "base64.h"
  47 #include "app_helper.h"
  48 #include "template.h"
  49
  50 using namespace nghttp2;
  51
  52 namespace shrpx {
  53
  54 namespace {
  55 constexpr size_t MAX_BUFFER_SIZE = 32_k;
  56 } // namespace
  57
  58 namespace {
  59 int on_stream_close_callback(nghttp2_session *session, int32_t stream_id,
  60                              uint32_t error_code, void *user_data) {
  61   auto upstream = static_cast<Http2Upstream *>(user_data);
  62   if (LOG_ENABLED(INFO)) {
  63     ULOG(INFO, upstream) << "Stream stream_id=" << stream_id
  64                          << " is being closed";
  65   }
  66
  67   auto downstream = static_cast<Downstream *>(
  68       nghttp2_session_get_stream_user_data(session, stream_id));
  69
  70   if (!downstream) {
  71     return 0;
  72   }
  73
  74   auto &req = downstream->request();
  75
  76   upstream->consume(stream_id, req.unconsumed_body_length);
  77
  78   req.unconsumed_body_length = 0;
  79
  80   if (downstream->get_request_state() == DownstreamState::CONNECT_FAIL) {
  81     upstream->remove_downstream(downstream);
  82     // downstream was deleted
  83
  84     return 0;
  85   }
  86
  87   if (downstream->can_detach_downstream_connection()) {
  88     // Keep-alive
  89     downstream->detach_downstream_connection();
  90   }
  91
  92   downstream->set_request_state(DownstreamState::STREAM_CLOSED);
  93
  94   // At this point, downstream read may be paused.
  95
  96   // If shrpx_downstream::push_request_headers() failed, the
  97   // error is handled here.
  98   upstream->remove_downstream(downstream);
  99   // downstream was deleted
 100
 101   // How to test this case? Request sufficient large download
 102   // and make client send RST_STREAM after it gets first DATA
 103   // frame chunk.
 104
 105   return 0;
 106 }
 107 } // namespace
 108
 109 int Http2Upstream::upgrade_upstream(HttpsUpstream *http) {
 110   int rv;
 111
 112   auto &balloc = http->get_downstream()->get_block_allocator();
 113
 114   auto http2_settings = http->get_downstream()->get_http2_settings();
 115   http2_settings = util::to_base64(balloc, http2_settings);
 116
 117   auto settings_payload = base64::decode(balloc, std::begin(http2_settings),
 118                                          std::end(http2_settings));
 119
 120   rv = nghttp2_session_upgrade2(
 121       session_, settings_payload.byte(), settings_payload.size(),
 122       http->get_downstream()->request().method == HTTP_HEAD, nullptr);
 123   if (rv != 0) {
 124     if (LOG_ENABLED(INFO)) {
 125       ULOG(INFO, this) << "nghttp2_session_upgrade() returned error: "
 126                        << nghttp2_strerror(rv);
 127     }
 128     return -1;
 129   }
 130   pre_upstream_.reset(http);
 131   auto downstream = http->pop_downstream();
 132   downstream->reset_upstream(this);
 133   downstream->set_stream_id(1);
 134   downstream->reset_upstream_rtimer();
 135   downstream->set_stream_id(1);
 136
 137   auto ptr = downstream.get();
 138
 139   nghttp2_session_set_stream_user_data(session_, 1, ptr);
 140   downstream_queue_.add_pending(std::move(downstream));
 141   downstream_queue_.mark_active(ptr);
 142
 143   // TODO This might not be necessary
 144   handler_->stop_read_timer();
 145
 146   if (LOG_ENABLED(INFO)) {
 147     ULOG(INFO, this) << "Connection upgraded to HTTP/2";
 148   }
 149
 150   return 0;
 151 }
 152
 153 void Http2Upstream::start_settings_timer() {
 154   ev_timer_start(handler_->get_loop(), &settings_timer_);
 155 }
 156
 157 void Http2Upstream::stop_settings_timer() {
 158   ev_timer_stop(handler_->get_loop(), &settings_timer_);
 159 }
 160
 161 namespace {
 162 int on_header_callback2(nghttp2_session *session, const nghttp2_frame *frame,
 163                         nghttp2_rcbuf *name, nghttp2_rcbuf *value,
 164                         uint8_t flags, void *user_data) {
 165   auto namebuf = nghttp2_rcbuf_get_buf(name);
 166   auto valuebuf = nghttp2_rcbuf_get_buf(value);
 167   auto config = get_config();
 168
 169   if (config->http2.upstream.debug.frame_debug) {
 170     verbose_on_header_callback(session, frame, namebuf.base, namebuf.len,
 171                                valuebuf.base, valuebuf.len, flags, user_data);
 172   }
 173   if (frame->hd.type != NGHTTP2_HEADERS) {
 174     return 0;
 175   }
 176   auto upstream = static_cast<Http2Upstream *>(user_data);
 177   auto downstream = static_cast<Downstream *>(
 178       nghttp2_session_get_stream_user_data(session, frame->hd.stream_id));
 179   if (!downstream) {
 180     return 0;
 181   }
 182
 183   auto &req = downstream->request();
 184
 185   auto &httpconf = config->http;
 186
 187   if (req.fs.buffer_size() + namebuf.len + valuebuf.len >
 188           httpconf.request_header_field_buffer ||
 189       req.fs.num_fields() >= httpconf.max_request_header_fields) {
 190     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
 191       return 0;
 192     }
 193
 194     if (LOG_ENABLED(INFO)) {
 195       ULOG(INFO, upstream) << "Too large or many header field size="
 196                            << req.fs.buffer_size() + namebuf.len + valuebuf.len
 197                            << ", num=" << req.fs.num_fields() + 1;
 198     }
 199
 200     // just ignore header fields if this is trailer part.
 201     if (frame->headers.cat == NGHTTP2_HCAT_HEADERS) {
 202       return 0;
 203     }
 204
 205     if (upstream->error_reply(downstream, 431) != 0) {
 206       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
 207     }
 208
 209     return 0;
 210   }
 211
 212   auto token = http2::lookup_token(namebuf.base, namebuf.len);
 213   auto no_index = flags & NGHTTP2_NV_FLAG_NO_INDEX;
 214
 215   downstream->add_rcbuf(name);
 216   downstream->add_rcbuf(value);
 217
 218   if (frame->headers.cat == NGHTTP2_HCAT_HEADERS) {
 219     // just store header fields for trailer part
 220     req.fs.add_trailer_token(StringRef{namebuf.base, namebuf.len},
 221                              StringRef{valuebuf.base, valuebuf.len}, no_index,
 222                              token);
 223     return 0;
 224   }
 225
 226   req.fs.add_header_token(StringRef{namebuf.base, namebuf.len},
 227                           StringRef{valuebuf.base, valuebuf.len}, no_index,
 228                           token);
 229   return 0;
 230 }
 231 } // namespace
 232
 233 namespace {
 234 int on_invalid_header_callback2(nghttp2_session *session,
 235                                 const nghttp2_frame *frame, nghttp2_rcbuf *name,
 236                                 nghttp2_rcbuf *value, uint8_t flags,
 237                                 void *user_data) {
 238   auto upstream = static_cast<Http2Upstream *>(user_data);
 239   auto downstream = static_cast<Downstream *>(
 240       nghttp2_session_get_stream_user_data(session, frame->hd.stream_id));
 241   if (!downstream) {
 242     return 0;
 243   }
 244
 245   if (LOG_ENABLED(INFO)) {
 246     auto namebuf = nghttp2_rcbuf_get_buf(name);
 247     auto valuebuf = nghttp2_rcbuf_get_buf(value);
 248
 249     ULOG(INFO, upstream) << "Invalid header field for stream_id="
 250                          << frame->hd.stream_id << ": name=["
 251                          << StringRef{namebuf.base, namebuf.len} << "], value=["
 252                          << StringRef{valuebuf.base, valuebuf.len} << "]";
 253   }
 254
 255   upstream->rst_stream(downstream, NGHTTP2_PROTOCOL_ERROR);
 256
 257   return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
 258 }
 259 } // namespace
 260
 261 namespace {
 262 int on_begin_headers_callback(nghttp2_session *session,
 263                               const nghttp2_frame *frame, void *user_data) {
 264   auto upstream = static_cast<Http2Upstream *>(user_data);
 265
 266   if (frame->headers.cat != NGHTTP2_HCAT_REQUEST) {
 267     return 0;
 268   }
 269   if (LOG_ENABLED(INFO)) {
 270     ULOG(INFO, upstream) << "Received upstream request HEADERS stream_id="
 271                          << frame->hd.stream_id;
 272   }
 273
 274   upstream->on_start_request(frame);
 275
 276   return 0;
 277 }
 278 } // namespace
 279
 280 void Http2Upstream::on_start_request(const nghttp2_frame *frame) {
 281   auto downstream = std::make_unique<Downstream>(this, handler_->get_mcpool(),
 282                                                  frame->hd.stream_id);
 283   nghttp2_session_set_stream_user_data(session_, frame->hd.stream_id,
 284                                        downstream.get());
 285
 286   downstream->reset_upstream_rtimer();
 287
 288   handler_->repeat_read_timer();
 289
 290   auto &req = downstream->request();
 291
 292   // Although, we deprecated minor version from HTTP/2, we supply
 293   // minor version 0 to use via header field in a conventional way.
 294   req.http_major = 2;
 295   req.http_minor = 0;
 296
 297   add_pending_downstream(std::move(downstream));
 298
 299   ++num_requests_;
 300
 301   auto config = get_config();
 302   auto &httpconf = config->http;
 303   if (httpconf.max_requests <= num_requests_) {
 304     start_graceful_shutdown();
 305   }
 306 }
 307
 308 int Http2Upstream::on_request_headers(Downstream *downstream,
 309                                       const nghttp2_frame *frame) {
 310   auto lgconf = log_config();
 311   lgconf->update_tstamp(std::chrono::system_clock::now());
 312   auto &req = downstream->request();
 313   req.tstamp = lgconf->tstamp;
 314
 315   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
 316     return 0;
 317   }
 318
 319   auto &nva = req.fs.headers();
 320
 321   if (LOG_ENABLED(INFO)) {
 322     std::stringstream ss;
 323     for (auto &nv : nva) {
 324       if (nv.name == "authorization") {
 325         ss << TTY_HTTP_HD << nv.name << TTY_RST << ": <redacted>\n";
 326         continue;
 327       }
 328       ss << TTY_HTTP_HD << nv.name << TTY_RST << ": " << nv.value << "\n";
 329     }
 330     ULOG(INFO, this) << "HTTP request headers. stream_id="
 331                      << downstream->get_stream_id() << "\n"
 332                      << ss.str();
 333   }
 334
 335   auto config = get_config();
 336   auto &dump = config->http2.upstream.debug.dump;
 337
 338   if (dump.request_header) {
 339     http2::dump_nv(dump.request_header, nva);
 340   }
 341
 342   auto content_length = req.fs.header(http2::HD_CONTENT_LENGTH);
 343   if (content_length) {
 344     // libnghttp2 guarantees this can be parsed
 345     req.fs.content_length = util::parse_uint(content_length->value);
 346   }
 347
 348   // presence of mandatory header fields are guaranteed by libnghttp2.
 349   auto authority = req.fs.header(http2::HD__AUTHORITY);
 350   auto path = req.fs.header(http2::HD__PATH);
 351   auto method = req.fs.header(http2::HD__METHOD);
 352   auto scheme = req.fs.header(http2::HD__SCHEME);
 353
 354   auto method_token = http2::lookup_method_token(method->value);
 355   if (method_token == -1) {
 356     if (error_reply(downstream, 501) != 0) {
 357       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
 358     }
 359     return 0;
 360   }
 361
 362   auto faddr = handler_->get_upstream_addr();
 363
 364   // For HTTP/2 proxy, we require :authority.
 365   if (method_token != HTTP_CONNECT && config->http2_proxy &&
 366       faddr->alt_mode == UpstreamAltMode::NONE && !authority) {
 367     rst_stream(downstream, NGHTTP2_PROTOCOL_ERROR);
 368     return 0;
 369   }
 370
 371   req.method = method_token;
 372   if (scheme) {
 373     req.scheme = scheme->value;
 374   }
 375
 376   // nghttp2 library guarantees either :authority or host exist
 377   if (!authority) {
 378     req.no_authority = true;
 379     authority = req.fs.header(http2::HD_HOST);
 380   }
 381
 382   if (authority) {
 383     req.authority = authority->value;
 384   }
 385
 386   if (path) {
 387     if (method_token == HTTP_OPTIONS &&
 388         path->value == StringRef::from_lit("*")) {
 389       // Server-wide OPTIONS request.  Path is empty.
 390     } else if (config->http2_proxy &&
 391                faddr->alt_mode == UpstreamAltMode::NONE) {
 392       req.path = path->value;
 393     } else {
 394       req.path = http2::rewrite_clean_path(downstream->get_block_allocator(),
 395                                            path->value);
 396     }
 397   }
 398
 399   auto connect_proto = req.fs.header(http2::HD__PROTOCOL);
 400   if (connect_proto) {
 401     if (connect_proto->value != "websocket") {
 402       if (error_reply(downstream, 400) != 0) {
 403         return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
 404       }
 405       return 0;
 406     }
 407     req.connect_proto = ConnectProto::WEBSOCKET;
 408   }
 409
 410   if (!(frame->hd.flags & NGHTTP2_FLAG_END_STREAM)) {
 411     req.http2_expect_body = true;
 412   } else if (req.fs.content_length == -1) {
 413     // If END_STREAM flag is set to HEADERS frame, we are sure that
 414     // content-length is 0.
 415     req.fs.content_length = 0;
 416   }
 417
 418   downstream->inspect_http2_request();
 419
 420   downstream->set_request_state(DownstreamState::HEADER_COMPLETE);
 421
 422 #ifdef HAVE_MRUBY
 423   auto upstream = downstream->get_upstream();
 424   auto handler = upstream->get_client_handler();
 425   auto worker = handler->get_worker();
 426   auto mruby_ctx = worker->get_mruby_context();
 427
 428   if (mruby_ctx->run_on_request_proc(downstream) != 0) {
 429     if (error_reply(downstream, 500) != 0) {
 430       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
 431     }
 432     return 0;
 433   }
 434 #endif // HAVE_MRUBY
 435
 436   if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
 437     downstream->disable_upstream_rtimer();
 438
 439     downstream->set_request_state(DownstreamState::MSG_COMPLETE);
 440   }
 441
 442   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
 443     return 0;
 444   }
 445
 446   start_downstream(downstream);
 447
 448   return 0;
 449 }
 450
 451 void Http2Upstream::start_downstream(Downstream *downstream) {
 452   if (downstream_queue_.can_activate(downstream->request().authority)) {
 453     initiate_downstream(downstream);
 454     return;
 455   }
 456
 457   downstream_queue_.mark_blocked(downstream);
 458 }
 459
 460 void Http2Upstream::initiate_downstream(Downstream *downstream) {
 461   int rv;
 462
 463   DownstreamConnection *dconn_ptr;
 464
 465   for (;;) {
 466     auto dconn = handler_->get_downstream_connection(rv, downstream);
 467     if (!dconn) {
 468       if (rv == SHRPX_ERR_TLS_REQUIRED) {
 469         rv = redirect_to_https(downstream);
 470       } else {
 471         rv = error_reply(downstream, 502);
 472       }
 473       if (rv != 0) {
 474         rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 475       }
 476
 477       downstream->set_request_state(DownstreamState::CONNECT_FAIL);
 478       downstream_queue_.mark_failure(downstream);
 479
 480       return;
 481     }
 482
 483 #ifdef HAVE_MRUBY
 484     dconn_ptr = dconn.get();
 485 #endif // HAVE_MRUBY
 486     rv = downstream->attach_downstream_connection(std::move(dconn));
 487     if (rv == 0) {
 488       break;
 489     }
 490   }
 491
 492 #ifdef HAVE_MRUBY
 493   const auto &group = dconn_ptr->get_downstream_addr_group();
 494   if (group) {
 495     const auto &mruby_ctx = group->shared_addr->mruby_ctx;
 496     if (mruby_ctx->run_on_request_proc(downstream) != 0) {
 497       if (error_reply(downstream, 500) != 0) {
 498         rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 499       }
 500
 501       downstream_queue_.mark_failure(downstream);
 502
 503       return;
 504     }
 505
 506     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
 507       return;
 508     }
 509   }
 510 #endif // HAVE_MRUBY
 511
 512   rv = downstream->push_request_headers();
 513   if (rv != 0) {
 514
 515     if (error_reply(downstream, 502) != 0) {
 516       rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 517     }
 518
 519     downstream_queue_.mark_failure(downstream);
 520
 521     return;
 522   }
 523
 524   downstream_queue_.mark_active(downstream);
 525
 526   auto &req = downstream->request();
 527   if (!req.http2_expect_body) {
 528     rv = downstream->end_upload_data();
 529     if (rv != 0) {
 530       rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 531     }
 532   }
 533
 534   return;
 535 }
 536
 537 namespace {
 538 int on_frame_recv_callback(nghttp2_session *session, const nghttp2_frame *frame,
 539                            void *user_data) {
 540   if (get_config()->http2.upstream.debug.frame_debug) {
 541     verbose_on_frame_recv_callback(session, frame, user_data);
 542   }
 543   auto upstream = static_cast<Http2Upstream *>(user_data);
 544   auto handler = upstream->get_client_handler();
 545
 546   switch (frame->hd.type) {
 547   case NGHTTP2_DATA: {
 548     auto downstream = static_cast<Downstream *>(
 549         nghttp2_session_get_stream_user_data(session, frame->hd.stream_id));
 550     if (!downstream) {
 551       return 0;
 552     }
 553
 554     if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
 555       downstream->disable_upstream_rtimer();
 556
 557       if (downstream->end_upload_data() != 0) {
 558         if (downstream->get_response_state() != DownstreamState::MSG_COMPLETE) {
 559           upstream->rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 560         }
 561       }
 562
 563       downstream->set_request_state(DownstreamState::MSG_COMPLETE);
 564     }
 565
 566     return 0;
 567   }
 568   case NGHTTP2_HEADERS: {
 569     auto downstream = static_cast<Downstream *>(
 570         nghttp2_session_get_stream_user_data(session, frame->hd.stream_id));
 571     if (!downstream) {
 572       return 0;
 573     }
 574
 575     if (frame->headers.cat == NGHTTP2_HCAT_REQUEST) {
 576       downstream->reset_upstream_rtimer();
 577
 578       handler->stop_read_timer();
 579
 580       return upstream->on_request_headers(downstream, frame);
 581     }
 582
 583     if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
 584       downstream->disable_upstream_rtimer();
 585
 586       if (downstream->end_upload_data() != 0) {
 587         if (downstream->get_response_state() != DownstreamState::MSG_COMPLETE) {
 588           upstream->rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 589         }
 590       }
 591
 592       downstream->set_request_state(DownstreamState::MSG_COMPLETE);
 593     }
 594
 595     return 0;
 596   }
 597   case NGHTTP2_SETTINGS:
 598     if ((frame->hd.flags & NGHTTP2_FLAG_ACK) == 0) {
 599       return 0;
 600     }
 601     upstream->stop_settings_timer();
 602     return 0;
 603   case NGHTTP2_GOAWAY:
 604     if (LOG_ENABLED(INFO)) {
 605       auto debug_data = util::ascii_dump(frame->goaway.opaque_data,
 606                                          frame->goaway.opaque_data_len);
 607
 608       ULOG(INFO, upstream) << "GOAWAY received: last-stream-id="
 609                            << frame->goaway.last_stream_id
 610                            << ", error_code=" << frame->goaway.error_code
 611                            << ", debug_data=" << debug_data;
 612     }
 613     return 0;
 614   default:
 615     return 0;
 616   }
 617 }
 618 } // namespace
 619
 620 namespace {
 621 int on_data_chunk_recv_callback(nghttp2_session *session, uint8_t flags,
 622                                 int32_t stream_id, const uint8_t *data,
 623                                 size_t len, void *user_data) {
 624   auto upstream = static_cast<Http2Upstream *>(user_data);
 625   auto downstream = static_cast<Downstream *>(
 626       nghttp2_session_get_stream_user_data(session, stream_id));
 627
 628   if (!downstream) {
 629     if (upstream->consume(stream_id, len) != 0) {
 630       return NGHTTP2_ERR_CALLBACK_FAILURE;
 631     }
 632
 633     return 0;
 634   }
 635
 636   downstream->reset_upstream_rtimer();
 637
 638   if (downstream->push_upload_data_chunk(data, len) != 0) {
 639     if (downstream->get_response_state() != DownstreamState::MSG_COMPLETE) {
 640       upstream->rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 641     }
 642
 643     if (upstream->consume(stream_id, len) != 0) {
 644       return NGHTTP2_ERR_CALLBACK_FAILURE;
 645     }
 646
 647     return 0;
 648   }
 649
 650   return 0;
 651 }
 652 } // namespace
 653
 654 namespace {
 655 int on_frame_send_callback(nghttp2_session *session, const nghttp2_frame *frame,
 656                            void *user_data) {
 657   if (get_config()->http2.upstream.debug.frame_debug) {
 658     verbose_on_frame_send_callback(session, frame, user_data);
 659   }
 660   auto upstream = static_cast<Http2Upstream *>(user_data);
 661   auto handler = upstream->get_client_handler();
 662
 663   switch (frame->hd.type) {
 664   case NGHTTP2_DATA:
 665   case NGHTTP2_HEADERS: {
 666     if ((frame->hd.flags & NGHTTP2_FLAG_END_STREAM) == 0) {
 667       return 0;
 668     }
 669     // RST_STREAM if request is still incomplete.
 670     auto stream_id = frame->hd.stream_id;
 671     auto downstream = static_cast<Downstream *>(
 672         nghttp2_session_get_stream_user_data(session, stream_id));
 673
 674     if (!downstream) {
 675       return 0;
 676     }
 677
 678     // For tunneling, issue RST_STREAM to finish the stream.
 679     if (downstream->get_upgraded() ||
 680         nghttp2_session_get_stream_remote_close(session, stream_id) == 0) {
 681       if (LOG_ENABLED(INFO)) {
 682         ULOG(INFO, upstream)
 683             << "Send RST_STREAM to "
 684             << (downstream->get_upgraded() ? "tunneled " : "")
 685             << "stream stream_id=" << downstream->get_stream_id()
 686             << " to finish off incomplete request";
 687       }
 688
 689       upstream->rst_stream(downstream, NGHTTP2_NO_ERROR);
 690     }
 691
 692     return 0;
 693   }
 694   case NGHTTP2_SETTINGS:
 695     if ((frame->hd.flags & NGHTTP2_FLAG_ACK) == 0) {
 696       upstream->start_settings_timer();
 697     }
 698     return 0;
 699   case NGHTTP2_PUSH_PROMISE: {
 700     auto promised_stream_id = frame->push_promise.promised_stream_id;
 701
 702     if (nghttp2_session_get_stream_user_data(session, promised_stream_id)) {
 703       // In case of push from backend, downstream object was already
 704       // created.
 705       return 0;
 706     }
 707
 708     auto promised_downstream = std::make_unique<Downstream>(
 709         upstream, handler->get_mcpool(), promised_stream_id);
 710     auto &req = promised_downstream->request();
 711
 712     // As long as we use nghttp2_session_mem_send(), setting stream
 713     // user data here should not fail.  This is because this callback
 714     // is called just after frame was serialized.  So no worries about
 715     // hanging Downstream.
 716     nghttp2_session_set_stream_user_data(session, promised_stream_id,
 717                                          promised_downstream.get());
 718
 719     promised_downstream->set_assoc_stream_id(frame->hd.stream_id);
 720     promised_downstream->disable_upstream_rtimer();
 721
 722     req.http_major = 2;
 723     req.http_minor = 0;
 724
 725     req.fs.content_length = 0;
 726     req.http2_expect_body = false;
 727
 728     auto &promised_balloc = promised_downstream->get_block_allocator();
 729
 730     for (size_t i = 0; i < frame->push_promise.nvlen; ++i) {
 731       auto &nv = frame->push_promise.nva[i];
 732
 733       auto name =
 734           make_string_ref(promised_balloc, StringRef{nv.name, nv.namelen});
 735       auto value =
 736           make_string_ref(promised_balloc, StringRef{nv.value, nv.valuelen});
 737
 738       auto token = http2::lookup_token(nv.name, nv.namelen);
 739       switch (token) {
 740       case http2::HD__METHOD:
 741         req.method = http2::lookup_method_token(value);
 742         break;
 743       case http2::HD__SCHEME:
 744         req.scheme = value;
 745         break;
 746       case http2::HD__AUTHORITY:
 747         req.authority = value;
 748         break;
 749       case http2::HD__PATH:
 750         req.path = http2::rewrite_clean_path(promised_balloc, value);
 751         break;
 752       }
 753       req.fs.add_header_token(name, value, nv.flags & NGHTTP2_NV_FLAG_NO_INDEX,
 754                               token);
 755     }
 756
 757     promised_downstream->inspect_http2_request();
 758
 759     promised_downstream->set_request_state(DownstreamState::MSG_COMPLETE);
 760
 761     // a bit weird but start_downstream() expects that given
 762     // downstream is in pending queue.
 763     auto ptr = promised_downstream.get();
 764     upstream->add_pending_downstream(std::move(promised_downstream));
 765
 766 #ifdef HAVE_MRUBY
 767     auto worker = handler->get_worker();
 768     auto mruby_ctx = worker->get_mruby_context();
 769
 770     if (mruby_ctx->run_on_request_proc(ptr) != 0) {
 771       if (upstream->error_reply(ptr, 500) != 0) {
 772         upstream->rst_stream(ptr, NGHTTP2_INTERNAL_ERROR);
 773         return 0;
 774       }
 775       return 0;
 776     }
 777 #endif // HAVE_MRUBY
 778
 779     upstream->start_downstream(ptr);
 780
 781     return 0;
 782   }
 783   case NGHTTP2_GOAWAY:
 784     if (LOG_ENABLED(INFO)) {
 785       auto debug_data = util::ascii_dump(frame->goaway.opaque_data,
 786                                          frame->goaway.opaque_data_len);
 787
 788       ULOG(INFO, upstream) << "Sending GOAWAY: last-stream-id="
 789                            << frame->goaway.last_stream_id
 790                            << ", error_code=" << frame->goaway.error_code
 791                            << ", debug_data=" << debug_data;
 792     }
 793     return 0;
 794   default:
 795     return 0;
 796   }
 797 }
 798 } // namespace
 799
 800 namespace {
 801 int on_frame_not_send_callback(nghttp2_session *session,
 802                                const nghttp2_frame *frame, int lib_error_code,
 803                                void *user_data) {
 804   auto upstream = static_cast<Http2Upstream *>(user_data);
 805   if (LOG_ENABLED(INFO)) {
 806     ULOG(INFO, upstream) << "Failed to send control frame type="
 807                          << static_cast<uint32_t>(frame->hd.type)
 808                          << ", lib_error_code=" << lib_error_code << ":"
 809                          << nghttp2_strerror(lib_error_code);
 810   }
 811   if (frame->hd.type == NGHTTP2_HEADERS &&
 812       lib_error_code != NGHTTP2_ERR_STREAM_CLOSED &&
 813       lib_error_code != NGHTTP2_ERR_STREAM_CLOSING) {
 814     // To avoid stream hanging around, issue RST_STREAM.
 815     auto downstream = static_cast<Downstream *>(
 816         nghttp2_session_get_stream_user_data(session, frame->hd.stream_id));
 817     if (downstream) {
 818       upstream->rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
 819     }
 820   }
 821   return 0;
 822 }
 823 } // namespace
 824
 825 namespace {
 826 constexpr auto PADDING = std::array<uint8_t, 256>{};
 827 } // namespace
 828
 829 namespace {
 830 int send_data_callback(nghttp2_session *session, nghttp2_frame *frame,
 831                        const uint8_t *framehd, size_t length,
 832                        nghttp2_data_source *source, void *user_data) {
 833   auto downstream = static_cast<Downstream *>(source->ptr);
 834   auto upstream = static_cast<Http2Upstream *>(downstream->get_upstream());
 835   auto body = downstream->get_response_buf();
 836
 837   auto wb = upstream->get_response_buf();
 838
 839   size_t padlen = 0;
 840
 841   wb->append(framehd, 9);
 842   if (frame->data.padlen > 0) {
 843     padlen = frame->data.padlen - 1;
 844     wb->append(static_cast<uint8_t>(padlen));
 845   }
 846
 847   body->remove(*wb, length);
 848
 849   wb->append(PADDING.data(), padlen);
 850
 851   if (body->rleft() == 0) {
 852     downstream->disable_upstream_wtimer();
 853   } else {
 854     downstream->reset_upstream_wtimer();
 855   }
 856
 857   if (length > 0 && downstream->resume_read(SHRPX_NO_BUFFER, length) != 0) {
 858     return NGHTTP2_ERR_CALLBACK_FAILURE;
 859   }
 860
 861   // We have to add length here, so that we can log this amount of
 862   // data transferred.
 863   downstream->response_sent_body_length += length;
 864
 865   auto max_buffer_size = upstream->get_max_buffer_size();
 866
 867   return wb->rleft() >= max_buffer_size ? NGHTTP2_ERR_PAUSE : 0;
 868 }
 869 } // namespace
 870
 871 namespace {
 872 uint32_t infer_upstream_rst_stream_error_code(uint32_t downstream_error_code) {
 873   // NGHTTP2_REFUSED_STREAM is important because it tells upstream
 874   // client to retry.
 875   switch (downstream_error_code) {
 876   case NGHTTP2_NO_ERROR:
 877   case NGHTTP2_REFUSED_STREAM:
 878     return downstream_error_code;
 879   default:
 880     return NGHTTP2_INTERNAL_ERROR;
 881   }
 882 }
 883 } // namespace
 884
 885 namespace {
 886 void settings_timeout_cb(struct ev_loop *loop, ev_timer *w, int revents) {
 887   auto upstream = static_cast<Http2Upstream *>(w->data);
 888   auto handler = upstream->get_client_handler();
 889   ULOG(INFO, upstream) << "SETTINGS timeout";
 890   if (upstream->terminate_session(NGHTTP2_SETTINGS_TIMEOUT) != 0) {
 891     delete handler;
 892     return;
 893   }
 894   handler->signal_write();
 895 }
 896 } // namespace
 897
 898 namespace {
 899 void shutdown_timeout_cb(struct ev_loop *loop, ev_timer *w, int revents) {
 900   auto upstream = static_cast<Http2Upstream *>(w->data);
 901   auto handler = upstream->get_client_handler();
 902   upstream->submit_goaway();
 903   handler->signal_write();
 904 }
 905 } // namespace
 906
 907 namespace {
 908 void prepare_cb(struct ev_loop *loop, ev_prepare *w, int revents) {
 909   auto upstream = static_cast<Http2Upstream *>(w->data);
 910   upstream->check_shutdown();
 911 }
 912 } // namespace
 913
 914 void Http2Upstream::submit_goaway() {
 915   auto last_stream_id = nghttp2_session_get_last_proc_stream_id(session_);
 916   nghttp2_submit_goaway(session_, NGHTTP2_FLAG_NONE, last_stream_id,
 917                         NGHTTP2_NO_ERROR, nullptr, 0);
 918 }
 919
 920 void Http2Upstream::check_shutdown() {
 921   auto worker = handler_->get_worker();
 922
 923   if (!worker->get_graceful_shutdown()) {
 924     return;
 925   }
 926
 927   ev_prepare_stop(handler_->get_loop(), &prep_);
 928
 929   start_graceful_shutdown();
 930 }
 931
 932 void Http2Upstream::start_graceful_shutdown() {
 933   int rv;
 934   if (ev_is_active(&shutdown_timer_)) {
 935     return;
 936   }
 937
 938   rv = nghttp2_submit_shutdown_notice(session_);
 939   if (rv != 0) {
 940     ULOG(FATAL, this) << "nghttp2_submit_shutdown_notice() failed: "
 941                       << nghttp2_strerror(rv);
 942     return;
 943   }
 944
 945   handler_->signal_write();
 946
 947   ev_timer_start(handler_->get_loop(), &shutdown_timer_);
 948 }
 949
 950 nghttp2_session_callbacks *create_http2_upstream_callbacks() {
 951   int rv;
 952   nghttp2_session_callbacks *callbacks;
 953
 954   rv = nghttp2_session_callbacks_new(&callbacks);
 955
 956   if (rv != 0) {
 957     return nullptr;
 958   }
 959
 960   nghttp2_session_callbacks_set_on_stream_close_callback(
 961       callbacks, on_stream_close_callback);
 962
 963   nghttp2_session_callbacks_set_on_frame_recv_callback(callbacks,
 964                                                        on_frame_recv_callback);
 965
 966   nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
 967       callbacks, on_data_chunk_recv_callback);
 968
 969   nghttp2_session_callbacks_set_on_frame_send_callback(callbacks,
 970                                                        on_frame_send_callback);
 971
 972   nghttp2_session_callbacks_set_on_frame_not_send_callback(
 973       callbacks, on_frame_not_send_callback);
 974
 975   nghttp2_session_callbacks_set_on_header_callback2(callbacks,
 976                                                     on_header_callback2);
 977
 978   nghttp2_session_callbacks_set_on_invalid_header_callback2(
 979       callbacks, on_invalid_header_callback2);
 980
 981   nghttp2_session_callbacks_set_on_begin_headers_callback(
 982       callbacks, on_begin_headers_callback);
 983
 984   nghttp2_session_callbacks_set_send_data_callback(callbacks,
 985                                                    send_data_callback);
 986
 987   auto config = get_config();
 988
 989   if (config->padding) {
 990     nghttp2_session_callbacks_set_select_padding_callback(
 991         callbacks, http::select_padding_callback);
 992   }
 993
 994   if (config->http2.upstream.debug.frame_debug) {
 995     nghttp2_session_callbacks_set_error_callback2(callbacks,
 996                                                   verbose_error_callback);
 997   }
 998
 999   return callbacks;
1000 }
1001
1002 namespace {
1003 size_t downstream_queue_size(Worker *worker) {
1004   auto &downstreamconf = *worker->get_downstream_config();
1005
1006   if (get_config()->http2_proxy) {
1007     return downstreamconf.connections_per_host;
1008   }
1009
1010   return downstreamconf.connections_per_frontend;
1011 }
1012 } // namespace
1013
1014 Http2Upstream::Http2Upstream(ClientHandler *handler)
1015     : wb_(handler->get_worker()->get_mcpool()),
1016       downstream_queue_(downstream_queue_size(handler->get_worker()),
1017                         !get_config()->http2_proxy),
1018       handler_(handler),
1019       session_(nullptr),
1020       max_buffer_size_(MAX_BUFFER_SIZE),
1021       num_requests_(0) {
1022   int rv;
1023
1024   auto config = get_config();
1025   auto &http2conf = config->http2;
1026
1027   auto faddr = handler_->get_upstream_addr();
1028
1029   rv =
1030       nghttp2_session_server_new2(&session_, http2conf.upstream.callbacks, this,
1031                                   faddr->alt_mode != UpstreamAltMode::NONE
1032                                       ? http2conf.upstream.alt_mode_option
1033                                       : http2conf.upstream.option);
1034
1035   assert(rv == 0);
1036
1037   flow_control_ = true;
1038
1039   // TODO Maybe call from outside?
1040   std::array<nghttp2_settings_entry, 4> entry;
1041   size_t nentry = 2;
1042
1043   entry[0].settings_id = NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS;
1044   entry[0].value = http2conf.upstream.max_concurrent_streams;
1045
1046   entry[1].settings_id = NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
1047   if (faddr->alt_mode != UpstreamAltMode::NONE) {
1048     entry[1].value = (1u << 31) - 1;
1049   } else {
1050     entry[1].value = http2conf.upstream.window_size;
1051   }
1052
1053   if (!config->http2_proxy) {
1054     entry[nentry].settings_id = NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL;
1055     entry[nentry].value = 1;
1056     ++nentry;
1057   }
1058
1059   if (http2conf.upstream.decoder_dynamic_table_size !=
1060       NGHTTP2_DEFAULT_HEADER_TABLE_SIZE) {
1061     entry[nentry].settings_id = NGHTTP2_SETTINGS_HEADER_TABLE_SIZE;
1062     entry[nentry].value = http2conf.upstream.decoder_dynamic_table_size;
1063     ++nentry;
1064   }
1065
1066   rv = nghttp2_submit_settings(session_, NGHTTP2_FLAG_NONE, entry.data(),
1067                                nentry);
1068   if (rv != 0) {
1069     ULOG(ERROR, this) << "nghttp2_submit_settings() returned error: "
1070                       << nghttp2_strerror(rv);
1071   }
1072
1073   auto window_size = faddr->alt_mode != UpstreamAltMode::NONE
1074                          ? std::numeric_limits<int32_t>::max()
1075                      : http2conf.upstream.optimize_window_size
1076                          ? std::min(http2conf.upstream.connection_window_size,
1077                                     NGHTTP2_INITIAL_CONNECTION_WINDOW_SIZE)
1078                          : http2conf.upstream.connection_window_size;
1079
1080   rv = nghttp2_session_set_local_window_size(session_, NGHTTP2_FLAG_NONE, 0,
1081                                              window_size);
1082
1083   if (rv != 0) {
1084     ULOG(ERROR, this)
1085         << "nghttp2_session_set_local_window_size() returned error: "
1086         << nghttp2_strerror(rv);
1087   }
1088
1089   // We wait for SETTINGS ACK at least 10 seconds.
1090   ev_timer_init(&settings_timer_, settings_timeout_cb,
1091                 http2conf.upstream.timeout.settings, 0.);
1092
1093   settings_timer_.data = this;
1094
1095   // timer for 2nd GOAWAY.  HTTP/2 spec recommend 1 RTT.  We wait for
1096   // 2 seconds.
1097   ev_timer_init(&shutdown_timer_, shutdown_timeout_cb, 2., 0);
1098   shutdown_timer_.data = this;
1099
1100   ev_prepare_init(&prep_, prepare_cb);
1101   prep_.data = this;
1102   ev_prepare_start(handler_->get_loop(), &prep_);
1103
1104 #if defined(TCP_INFO) && defined(TCP_NOTSENT_LOWAT)
1105   if (http2conf.upstream.optimize_write_buffer_size) {
1106     auto conn = handler_->get_connection();
1107     conn->tls_dyn_rec_warmup_threshold = 0;
1108
1109     uint32_t pollout_thres = 1;
1110     rv = setsockopt(conn->fd, IPPROTO_TCP, TCP_NOTSENT_LOWAT, &pollout_thres,
1111                     static_cast<socklen_t>(sizeof(pollout_thres)));
1112
1113     if (rv != 0) {
1114       if (LOG_ENABLED(INFO)) {
1115         auto error = errno;
1116         LOG(INFO) << "setsockopt(TCP_NOTSENT_LOWAT, " << pollout_thres
1117                   << ") failed: errno=" << error;
1118       }
1119     }
1120   }
1121 #endif // defined(TCP_INFO) && defined(TCP_NOTSENT_LOWAT)
1122
1123   handler_->reset_upstream_read_timeout(
1124       config->conn.upstream.timeout.http2_read);
1125
1126   handler_->signal_write();
1127 }
1128
1129 Http2Upstream::~Http2Upstream() {
1130   nghttp2_session_del(session_);
1131   ev_prepare_stop(handler_->get_loop(), &prep_);
1132   ev_timer_stop(handler_->get_loop(), &shutdown_timer_);
1133   ev_timer_stop(handler_->get_loop(), &settings_timer_);
1134 }
1135
1136 int Http2Upstream::on_read() {
1137   ssize_t rv = 0;
1138   auto rb = handler_->get_rb();
1139   auto rlimit = handler_->get_rlimit();
1140
1141   if (rb->rleft()) {
1142     rv = nghttp2_session_mem_recv(session_, rb->pos(), rb->rleft());
1143     if (rv < 0) {
1144       if (rv != NGHTTP2_ERR_BAD_CLIENT_MAGIC) {
1145         ULOG(ERROR, this) << "nghttp2_session_mem_recv() returned error: "
1146                           << nghttp2_strerror(rv);
1147       }
1148       return -1;
1149     }
1150
1151     // nghttp2_session_mem_recv should consume all input bytes on
1152     // success.
1153     assert(static_cast<size_t>(rv) == rb->rleft());
1154     rb->reset();
1155     rlimit->startw();
1156   }
1157
1158   if (nghttp2_session_want_read(session_) == 0 &&
1159       nghttp2_session_want_write(session_) == 0 && wb_.rleft() == 0) {
1160     if (LOG_ENABLED(INFO)) {
1161       ULOG(INFO, this) << "No more read/write for this HTTP2 session";
1162     }
1163     return -1;
1164   }
1165
1166   handler_->signal_write();
1167   return 0;
1168 }
1169
1170 // After this function call, downstream may be deleted.
1171 int Http2Upstream::on_write() {
1172   int rv;
1173   auto config = get_config();
1174   auto &http2conf = config->http2;
1175
1176   if ((http2conf.upstream.optimize_write_buffer_size ||
1177        http2conf.upstream.optimize_window_size) &&
1178       handler_->get_ssl()) {
1179     auto conn = handler_->get_connection();
1180     TCPHint hint;
1181     rv = conn->get_tcp_hint(&hint);
1182     if (rv == 0) {
1183       if (http2conf.upstream.optimize_write_buffer_size) {
1184         max_buffer_size_ = std::min(MAX_BUFFER_SIZE, hint.write_buffer_size);
1185       }
1186
1187       if (http2conf.upstream.optimize_window_size) {
1188         auto faddr = handler_->get_upstream_addr();
1189         if (faddr->alt_mode == UpstreamAltMode::NONE) {
1190           auto window_size = std::min(http2conf.upstream.connection_window_size,
1191                                       static_cast<int32_t>(hint.rwin * 2));
1192
1193           rv = nghttp2_session_set_local_window_size(
1194               session_, NGHTTP2_FLAG_NONE, 0, window_size);
1195           if (rv != 0) {
1196             if (LOG_ENABLED(INFO)) {
1197               ULOG(INFO, this)
1198                   << "nghttp2_session_set_local_window_size() with window_size="
1199                   << window_size << " failed: " << nghttp2_strerror(rv);
1200             }
1201           }
1202         }
1203       }
1204     }
1205   }
1206
1207   for (;;) {
1208     if (wb_.rleft() >= max_buffer_size_) {
1209       return 0;
1210     }
1211
1212     const uint8_t *data;
1213     auto datalen = nghttp2_session_mem_send(session_, &data);
1214
1215     if (datalen < 0) {
1216       ULOG(ERROR, this) << "nghttp2_session_mem_send() returned error: "
1217                         << nghttp2_strerror(datalen);
1218       return -1;
1219     }
1220     if (datalen == 0) {
1221       break;
1222     }
1223     wb_.append(data, datalen);
1224   }
1225
1226   if (nghttp2_session_want_read(session_) == 0 &&
1227       nghttp2_session_want_write(session_) == 0 && wb_.rleft() == 0) {
1228     if (LOG_ENABLED(INFO)) {
1229       ULOG(INFO, this) << "No more read/write for this HTTP2 session";
1230     }
1231     return -1;
1232   }
1233
1234   return 0;
1235 }
1236
1237 ClientHandler *Http2Upstream::get_client_handler() const { return handler_; }
1238
1239 int Http2Upstream::downstream_read(DownstreamConnection *dconn) {
1240   auto downstream = dconn->get_downstream();
1241
1242   if (downstream->get_response_state() == DownstreamState::MSG_RESET) {
1243     // The downstream stream was reset (canceled). In this case,
1244     // RST_STREAM to the upstream and delete downstream connection
1245     // here. Deleting downstream will be taken place at
1246     // on_stream_close_callback.
1247     rst_stream(downstream,
1248                infer_upstream_rst_stream_error_code(
1249                    downstream->get_response_rst_stream_error_code()));
1250     downstream->pop_downstream_connection();
1251     // dconn was deleted
1252     dconn = nullptr;
1253   } else if (downstream->get_response_state() ==
1254              DownstreamState::MSG_BAD_HEADER) {
1255     if (error_reply(downstream, 502) != 0) {
1256       return -1;
1257     }
1258     downstream->pop_downstream_connection();
1259     // dconn was deleted
1260     dconn = nullptr;
1261   } else {
1262     auto rv = downstream->on_read();
1263     if (rv == SHRPX_ERR_EOF) {
1264       if (downstream->get_request_header_sent()) {
1265         return downstream_eof(dconn);
1266       }
1267       return SHRPX_ERR_RETRY;
1268     }
1269     if (rv == SHRPX_ERR_DCONN_CANCELED) {
1270       downstream->pop_downstream_connection();
1271       handler_->signal_write();
1272       return 0;
1273     }
1274     if (rv != 0) {
1275       if (rv != SHRPX_ERR_NETWORK) {
1276         if (LOG_ENABLED(INFO)) {
1277           DCLOG(INFO, dconn) << "HTTP parser failure";
1278         }
1279       }
1280       return downstream_error(dconn, Downstream::EVENT_ERROR);
1281     }
1282
1283     if (downstream->can_detach_downstream_connection()) {
1284       // Keep-alive
1285       downstream->detach_downstream_connection();
1286     }
1287   }
1288
1289   handler_->signal_write();
1290
1291   // At this point, downstream may be deleted.
1292
1293   return 0;
1294 }
1295
1296 int Http2Upstream::downstream_write(DownstreamConnection *dconn) {
1297   int rv;
1298   rv = dconn->on_write();
1299   if (rv == SHRPX_ERR_NETWORK) {
1300     return downstream_error(dconn, Downstream::EVENT_ERROR);
1301   }
1302   if (rv != 0) {
1303     return rv;
1304   }
1305   return 0;
1306 }
1307
1308 int Http2Upstream::downstream_eof(DownstreamConnection *dconn) {
1309   auto downstream = dconn->get_downstream();
1310
1311   if (LOG_ENABLED(INFO)) {
1312     DCLOG(INFO, dconn) << "EOF. stream_id=" << downstream->get_stream_id();
1313   }
1314
1315   // Delete downstream connection. If we don't delete it here, it will
1316   // be pooled in on_stream_close_callback.
1317   downstream->pop_downstream_connection();
1318   // dconn was deleted
1319   dconn = nullptr;
1320   // downstream wil be deleted in on_stream_close_callback.
1321   if (downstream->get_response_state() == DownstreamState::HEADER_COMPLETE) {
1322     // Server may indicate the end of the request by EOF
1323     if (LOG_ENABLED(INFO)) {
1324       ULOG(INFO, this) << "Downstream body was ended by EOF";
1325     }
1326     downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1327
1328     // For tunneled connection, MSG_COMPLETE signals
1329     // downstream_data_read_callback to send RST_STREAM after pending
1330     // response body is sent. This is needed to ensure that RST_STREAM
1331     // is sent after all pending data are sent.
1332     on_downstream_body_complete(downstream);
1333   } else if (downstream->get_response_state() !=
1334              DownstreamState::MSG_COMPLETE) {
1335     // If stream was not closed, then we set MSG_COMPLETE and let
1336     // on_stream_close_callback delete downstream.
1337     if (error_reply(downstream, 502) != 0) {
1338       return -1;
1339     }
1340   }
1341   handler_->signal_write();
1342   // At this point, downstream may be deleted.
1343   return 0;
1344 }
1345
1346 int Http2Upstream::downstream_error(DownstreamConnection *dconn, int events) {
1347   auto downstream = dconn->get_downstream();
1348
1349   if (LOG_ENABLED(INFO)) {
1350     if (events & Downstream::EVENT_ERROR) {
1351       DCLOG(INFO, dconn) << "Downstream network/general error";
1352     } else {
1353       DCLOG(INFO, dconn) << "Timeout";
1354     }
1355     if (downstream->get_upgraded()) {
1356       DCLOG(INFO, dconn) << "Note: this is tunnel connection";
1357     }
1358   }
1359
1360   // Delete downstream connection. If we don't delete it here, it will
1361   // be pooled in on_stream_close_callback.
1362   downstream->pop_downstream_connection();
1363   // dconn was deleted
1364   dconn = nullptr;
1365
1366   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1367     // For SSL tunneling, we issue RST_STREAM. For other types of
1368     // stream, we don't have to do anything since response was
1369     // complete.
1370     if (downstream->get_upgraded()) {
1371       rst_stream(downstream, NGHTTP2_NO_ERROR);
1372     }
1373   } else {
1374     if (downstream->get_response_state() == DownstreamState::HEADER_COMPLETE) {
1375       if (downstream->get_upgraded()) {
1376         on_downstream_body_complete(downstream);
1377       } else {
1378         rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
1379       }
1380     } else {
1381       unsigned int status;
1382       if (events & Downstream::EVENT_TIMEOUT) {
1383         if (downstream->get_request_header_sent()) {
1384           status = 504;
1385         } else {
1386           status = 408;
1387         }
1388       } else {
1389         status = 502;
1390       }
1391       if (error_reply(downstream, status) != 0) {
1392         return -1;
1393       }
1394     }
1395     downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1396   }
1397   handler_->signal_write();
1398   // At this point, downstream may be deleted.
1399   return 0;
1400 }
1401
1402 int Http2Upstream::rst_stream(Downstream *downstream, uint32_t error_code) {
1403   if (LOG_ENABLED(INFO)) {
1404     ULOG(INFO, this) << "RST_STREAM stream_id=" << downstream->get_stream_id()
1405                      << " with error_code=" << error_code;
1406   }
1407   int rv;
1408   rv = nghttp2_submit_rst_stream(session_, NGHTTP2_FLAG_NONE,
1409                                  downstream->get_stream_id(), error_code);
1410   if (rv < NGHTTP2_ERR_FATAL) {
1411     ULOG(FATAL, this) << "nghttp2_submit_rst_stream() failed: "
1412                       << nghttp2_strerror(rv);
1413     return -1;
1414   }
1415   return 0;
1416 }
1417
1418 int Http2Upstream::terminate_session(uint32_t error_code) {
1419   int rv;
1420   rv = nghttp2_session_terminate_session(session_, error_code);
1421   if (rv != 0) {
1422     return -1;
1423   }
1424   return 0;
1425 }
1426
1427 namespace {
1428 ssize_t downstream_data_read_callback(nghttp2_session *session,
1429                                       int32_t stream_id, uint8_t *buf,
1430                                       size_t length, uint32_t *data_flags,
1431                                       nghttp2_data_source *source,
1432                                       void *user_data) {
1433   int rv;
1434   auto downstream = static_cast<Downstream *>(source->ptr);
1435   auto body = downstream->get_response_buf();
1436   assert(body);
1437   auto upstream = static_cast<Http2Upstream *>(user_data);
1438
1439   const auto &resp = downstream->response();
1440
1441   auto nread = std::min(body->rleft(), length);
1442
1443   auto max_buffer_size = upstream->get_max_buffer_size();
1444
1445   auto buffer = upstream->get_response_buf();
1446
1447   if (max_buffer_size <
1448       std::min(nread, static_cast<size_t>(256)) + 9 + buffer->rleft()) {
1449     if (LOG_ENABLED(INFO)) {
1450       ULOG(INFO, upstream) << "Buffer is almost full.  Skip write DATA";
1451     }
1452     return NGHTTP2_ERR_PAUSE;
1453   }
1454
1455   nread = std::min(nread, max_buffer_size - 9 - buffer->rleft());
1456
1457   auto body_empty = body->rleft() == nread;
1458
1459   *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
1460
1461   if (body_empty &&
1462       downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1463
1464     *data_flags |= NGHTTP2_DATA_FLAG_EOF;
1465
1466     if (!downstream->get_upgraded()) {
1467       const auto &trailers = resp.fs.trailers();
1468       if (!trailers.empty()) {
1469         std::vector<nghttp2_nv> nva;
1470         nva.reserve(trailers.size());
1471         http2::copy_headers_to_nva_nocopy(nva, trailers, http2::HDOP_STRIP_ALL);
1472         if (!nva.empty()) {
1473           rv = nghttp2_submit_trailer(session, stream_id, nva.data(),
1474                                       nva.size());
1475           if (rv != 0) {
1476             if (nghttp2_is_fatal(rv)) {
1477               return NGHTTP2_ERR_CALLBACK_FAILURE;
1478             }
1479           } else {
1480             *data_flags |= NGHTTP2_DATA_FLAG_NO_END_STREAM;
1481           }
1482         }
1483       }
1484     }
1485   }
1486
1487   if (nread == 0 && ((*data_flags) & NGHTTP2_DATA_FLAG_EOF) == 0) {
1488     downstream->disable_upstream_wtimer();
1489     return NGHTTP2_ERR_DEFERRED;
1490   }
1491
1492   return nread;
1493 }
1494 } // namespace
1495
1496 int Http2Upstream::send_reply(Downstream *downstream, const uint8_t *body,
1497                               size_t bodylen) {
1498   int rv;
1499
1500   nghttp2_data_provider data_prd, *data_prd_ptr = nullptr;
1501
1502   if (bodylen) {
1503     data_prd.source.ptr = downstream;
1504     data_prd.read_callback = downstream_data_read_callback;
1505     data_prd_ptr = &data_prd;
1506   }
1507
1508   const auto &resp = downstream->response();
1509   auto config = get_config();
1510   auto &httpconf = config->http;
1511
1512   auto &balloc = downstream->get_block_allocator();
1513
1514   const auto &headers = resp.fs.headers();
1515   auto nva = std::vector<nghttp2_nv>();
1516   // 2 for :status and server
1517   nva.reserve(2 + headers.size() + httpconf.add_response_headers.size());
1518
1519   auto response_status = http2::stringify_status(balloc, resp.http_status);
1520
1521   nva.push_back(http2::make_nv_ls_nocopy(":status", response_status));
1522
1523   for (auto &kv : headers) {
1524     if (kv.name.empty() || kv.name[0] == ':') {
1525       continue;
1526     }
1527     switch (kv.token) {
1528     case http2::HD_CONNECTION:
1529     case http2::HD_KEEP_ALIVE:
1530     case http2::HD_PROXY_CONNECTION:
1531     case http2::HD_TE:
1532     case http2::HD_TRANSFER_ENCODING:
1533     case http2::HD_UPGRADE:
1534       continue;
1535     }
1536     nva.push_back(http2::make_nv_nocopy(kv.name, kv.value, kv.no_index));
1537   }
1538
1539   if (!resp.fs.header(http2::HD_SERVER)) {
1540     nva.push_back(http2::make_nv_ls_nocopy("server", config->http.server_name));
1541   }
1542
1543   for (auto &p : httpconf.add_response_headers) {
1544     nva.push_back(http2::make_nv_nocopy(p.name, p.value));
1545   }
1546
1547   rv = nghttp2_submit_response(session_, downstream->get_stream_id(),
1548                                nva.data(), nva.size(), data_prd_ptr);
1549   if (nghttp2_is_fatal(rv)) {
1550     ULOG(FATAL, this) << "nghttp2_submit_response() failed: "
1551                       << nghttp2_strerror(rv);
1552     return -1;
1553   }
1554
1555   auto buf = downstream->get_response_buf();
1556
1557   buf->append(body, bodylen);
1558
1559   downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1560
1561   if (data_prd_ptr) {
1562     downstream->reset_upstream_wtimer();
1563   }
1564
1565   return 0;
1566 }
1567
1568 int Http2Upstream::error_reply(Downstream *downstream,
1569                                unsigned int status_code) {
1570   int rv;
1571   auto &resp = downstream->response();
1572
1573   auto &balloc = downstream->get_block_allocator();
1574
1575   auto html = http::create_error_html(balloc, status_code);
1576   resp.http_status = status_code;
1577   auto body = downstream->get_response_buf();
1578   body->append(html);
1579   downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1580
1581   nghttp2_data_provider data_prd;
1582   data_prd.source.ptr = downstream;
1583   data_prd.read_callback = downstream_data_read_callback;
1584
1585   auto lgconf = log_config();
1586   lgconf->update_tstamp(std::chrono::system_clock::now());
1587
1588   auto response_status = http2::stringify_status(balloc, status_code);
1589   auto content_length = util::make_string_ref_uint(balloc, html.size());
1590   auto date = make_string_ref(balloc, lgconf->tstamp->time_http);
1591
1592   auto nva = std::array<nghttp2_nv, 5>{
1593       {http2::make_nv_ls_nocopy(":status", response_status),
1594        http2::make_nv_ll("content-type", "text/html; charset=UTF-8"),
1595        http2::make_nv_ls_nocopy("server", get_config()->http.server_name),
1596        http2::make_nv_ls_nocopy("content-length", content_length),
1597        http2::make_nv_ls_nocopy("date", date)}};
1598
1599   rv = nghttp2_submit_response(session_, downstream->get_stream_id(),
1600                                nva.data(), nva.size(), &data_prd);
1601   if (rv < NGHTTP2_ERR_FATAL) {
1602     ULOG(FATAL, this) << "nghttp2_submit_response() failed: "
1603                       << nghttp2_strerror(rv);
1604     return -1;
1605   }
1606
1607   downstream->reset_upstream_wtimer();
1608
1609   return 0;
1610 }
1611
1612 void Http2Upstream::add_pending_downstream(
1613     std::unique_ptr<Downstream> downstream) {
1614   downstream_queue_.add_pending(std::move(downstream));
1615 }
1616
1617 void Http2Upstream::remove_downstream(Downstream *downstream) {
1618   if (downstream->accesslog_ready()) {
1619     handler_->write_accesslog(downstream);
1620   }
1621
1622   nghttp2_session_set_stream_user_data(session_, downstream->get_stream_id(),
1623                                        nullptr);
1624
1625   auto next_downstream = downstream_queue_.remove_and_get_blocked(downstream);
1626
1627   if (next_downstream) {
1628     initiate_downstream(next_downstream);
1629   }
1630
1631   if (downstream_queue_.get_downstreams() == nullptr) {
1632     // There is no downstream at the moment.  Start idle timer now.
1633     handler_->repeat_read_timer();
1634   }
1635 }
1636
1637 // WARNING: Never call directly or indirectly nghttp2_session_send or
1638 // nghttp2_session_recv. These calls may delete downstream.
1639 int Http2Upstream::on_downstream_header_complete(Downstream *downstream) {
1640   int rv;
1641
1642   const auto &req = downstream->request();
1643   auto &resp = downstream->response();
1644
1645   auto &balloc = downstream->get_block_allocator();
1646
1647   if (LOG_ENABLED(INFO)) {
1648     if (downstream->get_non_final_response()) {
1649       DLOG(INFO, downstream) << "HTTP non-final response header";
1650     } else {
1651       DLOG(INFO, downstream) << "HTTP response header completed";
1652     }
1653   }
1654
1655   auto config = get_config();
1656   auto &httpconf = config->http;
1657
1658   if (!config->http2_proxy && !httpconf.no_location_rewrite) {
1659     downstream->rewrite_location_response_header(req.scheme);
1660   }
1661
1662 #ifdef HAVE_MRUBY
1663   if (!downstream->get_non_final_response()) {
1664     auto dconn = downstream->get_downstream_connection();
1665     const auto &group = dconn->get_downstream_addr_group();
1666     if (group) {
1667       const auto &dmruby_ctx = group->shared_addr->mruby_ctx;
1668
1669       if (dmruby_ctx->run_on_response_proc(downstream) != 0) {
1670         if (error_reply(downstream, 500) != 0) {
1671           return -1;
1672         }
1673         // Returning -1 will signal deletion of dconn.
1674         return -1;
1675       }
1676
1677       if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1678         return -1;
1679       }
1680     }
1681
1682     auto worker = handler_->get_worker();
1683     auto mruby_ctx = worker->get_mruby_context();
1684
1685     if (mruby_ctx->run_on_response_proc(downstream) != 0) {
1686       if (error_reply(downstream, 500) != 0) {
1687         return -1;
1688       }
1689       // Returning -1 will signal deletion of dconn.
1690       return -1;
1691     }
1692
1693     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1694       return -1;
1695     }
1696   }
1697 #endif // HAVE_MRUBY
1698
1699   auto &http2conf = config->http2;
1700
1701   // We need some conditions that must be fulfilled to initiate server
1702   // push.
1703   //
1704   // * Server push is disabled for http2 proxy or client proxy, since
1705   //   incoming headers are mixed origins.  We don't know how to
1706   //   reliably determine the authority yet.
1707   //
1708   // * We need non-final response or 200 response code for associated
1709   //   resource.  This is too restrictive, we will review this later.
1710   //
1711   // * We requires GET or POST for associated resource.  Probably we
1712   //   don't want to push for HEAD request.  Not sure other methods
1713   //   are also eligible for push.
1714   if (!http2conf.no_server_push &&
1715       nghttp2_session_get_remote_settings(session_,
1716                                           NGHTTP2_SETTINGS_ENABLE_PUSH) == 1 &&
1717       !config->http2_proxy && (downstream->get_stream_id() % 2) &&
1718       resp.fs.header(http2::HD_LINK) &&
1719       (downstream->get_non_final_response() || resp.http_status == 200) &&
1720       (req.method == HTTP_GET || req.method == HTTP_POST)) {
1721
1722     if (prepare_push_promise(downstream) != 0) {
1723       // Continue to send response even if push was failed.
1724     }
1725   }
1726
1727   auto nva = std::vector<nghttp2_nv>();
1728   // 6 means :status and possible server, via, x-http2-push, alt-svc,
1729   // and set-cookie (for affinity cookie) header field.
1730   nva.reserve(resp.fs.headers().size() + 6 +
1731               httpconf.add_response_headers.size());
1732
1733   if (downstream->get_non_final_response()) {
1734     auto response_status = http2::stringify_status(balloc, resp.http_status);
1735
1736     nva.push_back(http2::make_nv_ls_nocopy(":status", response_status));
1737
1738     http2::copy_headers_to_nva_nocopy(nva, resp.fs.headers(),
1739                                       http2::HDOP_STRIP_ALL);
1740
1741     if (LOG_ENABLED(INFO)) {
1742       log_response_headers(downstream, nva);
1743     }
1744
1745     rv = nghttp2_submit_headers(session_, NGHTTP2_FLAG_NONE,
1746                                 downstream->get_stream_id(), nullptr,
1747                                 nva.data(), nva.size(), nullptr);
1748
1749     resp.fs.clear_headers();
1750
1751     if (rv != 0) {
1752       ULOG(FATAL, this) << "nghttp2_submit_headers() failed";
1753       return -1;
1754     }
1755
1756     return 0;
1757   }
1758
1759   auto striphd_flags = http2::HDOP_STRIP_ALL & ~http2::HDOP_STRIP_VIA;
1760   StringRef response_status;
1761
1762   if (req.connect_proto == ConnectProto::WEBSOCKET && resp.http_status == 101) {
1763     response_status = http2::stringify_status(balloc, 200);
1764     striphd_flags |= http2::HDOP_STRIP_SEC_WEBSOCKET_ACCEPT;
1765   } else {
1766     response_status = http2::stringify_status(balloc, resp.http_status);
1767   }
1768
1769   nva.push_back(http2::make_nv_ls_nocopy(":status", response_status));
1770
1771   http2::copy_headers_to_nva_nocopy(nva, resp.fs.headers(), striphd_flags);
1772
1773   if (!config->http2_proxy && !httpconf.no_server_rewrite) {
1774     nva.push_back(http2::make_nv_ls_nocopy("server", httpconf.server_name));
1775   } else {
1776     auto server = resp.fs.header(http2::HD_SERVER);
1777     if (server) {
1778       nva.push_back(http2::make_nv_ls_nocopy("server", (*server).value));
1779     }
1780   }
1781
1782   if (!req.regular_connect_method() || !downstream->get_upgraded()) {
1783     auto affinity_cookie = downstream->get_affinity_cookie_to_send();
1784     if (affinity_cookie) {
1785       auto dconn = downstream->get_downstream_connection();
1786       assert(dconn);
1787       auto &group = dconn->get_downstream_addr_group();
1788       auto &shared_addr = group->shared_addr;
1789       auto &cookieconf = shared_addr->affinity.cookie;
1790       auto secure =
1791           http::require_cookie_secure_attribute(cookieconf.secure, req.scheme);
1792       auto cookie_str = http::create_affinity_cookie(
1793           balloc, cookieconf.name, affinity_cookie, cookieconf.path, secure);
1794       nva.push_back(http2::make_nv_ls_nocopy("set-cookie", cookie_str));
1795     }
1796   }
1797
1798   if (!resp.fs.header(http2::HD_ALT_SVC)) {
1799     // We won't change or alter alt-svc from backend for now
1800     if (!httpconf.http2_altsvc_header_value.empty()) {
1801       nva.push_back(http2::make_nv_ls_nocopy(
1802           "alt-svc", httpconf.http2_altsvc_header_value));
1803     }
1804   }
1805
1806   auto via = resp.fs.header(http2::HD_VIA);
1807   if (httpconf.no_via) {
1808     if (via) {
1809       nva.push_back(http2::make_nv_ls_nocopy("via", (*via).value));
1810     }
1811   } else {
1812     // we don't create more than 16 bytes in
1813     // http::create_via_header_value.
1814     size_t len = 16;
1815     if (via) {
1816       len += via->value.size() + 2;
1817     }
1818
1819     auto iov = make_byte_ref(balloc, len + 1);
1820     auto p = iov.base;
1821     if (via) {
1822       p = std::copy(std::begin(via->value), std::end(via->value), p);
1823       p = util::copy_lit(p, ", ");
1824     }
1825     p = http::create_via_header_value(p, resp.http_major, resp.http_minor);
1826     *p = '\0';
1827
1828     nva.push_back(http2::make_nv_ls_nocopy("via", StringRef{iov.base, p}));
1829   }
1830
1831   for (auto &p : httpconf.add_response_headers) {
1832     nva.push_back(http2::make_nv_nocopy(p.name, p.value));
1833   }
1834
1835   if (downstream->get_stream_id() % 2 == 0) {
1836     // This header field is basically for human on client side to
1837     // figure out that the resource is pushed.
1838     nva.push_back(http2::make_nv_ll("x-http2-push", "1"));
1839   }
1840
1841   if (LOG_ENABLED(INFO)) {
1842     log_response_headers(downstream, nva);
1843   }
1844
1845   if (http2conf.upstream.debug.dump.response_header) {
1846     http2::dump_nv(http2conf.upstream.debug.dump.response_header, nva.data(),
1847                    nva.size());
1848   }
1849
1850   nghttp2_data_provider data_prd;
1851   data_prd.source.ptr = downstream;
1852   data_prd.read_callback = downstream_data_read_callback;
1853
1854   nghttp2_data_provider *data_prdptr;
1855
1856   if (downstream->expect_response_body() ||
1857       downstream->expect_response_trailer()) {
1858     data_prdptr = &data_prd;
1859   } else {
1860     data_prdptr = nullptr;
1861   }
1862
1863   rv = nghttp2_submit_response(session_, downstream->get_stream_id(),
1864                                nva.data(), nva.size(), data_prdptr);
1865   if (rv != 0) {
1866     ULOG(FATAL, this) << "nghttp2_submit_response() failed";
1867     return -1;
1868   }
1869
1870   if (data_prdptr) {
1871     downstream->reset_upstream_wtimer();
1872   }
1873
1874   return 0;
1875 }
1876
1877 // WARNING: Never call directly or indirectly nghttp2_session_send or
1878 // nghttp2_session_recv. These calls may delete downstream.
1879 int Http2Upstream::on_downstream_body(Downstream *downstream,
1880                                       const uint8_t *data, size_t len,
1881                                       bool flush) {
1882   auto body = downstream->get_response_buf();
1883   body->append(data, len);
1884
1885   if (flush) {
1886     nghttp2_session_resume_data(session_, downstream->get_stream_id());
1887
1888     downstream->ensure_upstream_wtimer();
1889   }
1890
1891   return 0;
1892 }
1893
1894 // WARNING: Never call directly or indirectly nghttp2_session_send or
1895 // nghttp2_session_recv. These calls may delete downstream.
1896 int Http2Upstream::on_downstream_body_complete(Downstream *downstream) {
1897   if (LOG_ENABLED(INFO)) {
1898     DLOG(INFO, downstream) << "HTTP response completed";
1899   }
1900
1901   auto &resp = downstream->response();
1902
1903   if (!downstream->validate_response_recv_body_length()) {
1904     rst_stream(downstream, NGHTTP2_PROTOCOL_ERROR);
1905     resp.connection_close = true;
1906     return 0;
1907   }
1908
1909   nghttp2_session_resume_data(session_, downstream->get_stream_id());
1910   downstream->ensure_upstream_wtimer();
1911
1912   return 0;
1913 }
1914
1915 bool Http2Upstream::get_flow_control() const { return flow_control_; }
1916
1917 void Http2Upstream::pause_read(IOCtrlReason reason) {}
1918
1919 int Http2Upstream::resume_read(IOCtrlReason reason, Downstream *downstream,
1920                                size_t consumed) {
1921   if (get_flow_control()) {
1922     if (consume(downstream->get_stream_id(), consumed) != 0) {
1923       return -1;
1924     }
1925
1926     auto &req = downstream->request();
1927
1928     req.consume(consumed);
1929   }
1930
1931   handler_->signal_write();
1932   return 0;
1933 }
1934
1935 int Http2Upstream::on_downstream_abort_request(Downstream *downstream,
1936                                                unsigned int status_code) {
1937   int rv;
1938
1939   rv = error_reply(downstream, status_code);
1940
1941   if (rv != 0) {
1942     return -1;
1943   }
1944
1945   handler_->signal_write();
1946   return 0;
1947 }
1948
1949 int Http2Upstream::on_downstream_abort_request_with_https_redirect(
1950     Downstream *downstream) {
1951   int rv;
1952
1953   rv = redirect_to_https(downstream);
1954   if (rv != 0) {
1955     return -1;
1956   }
1957
1958   handler_->signal_write();
1959   return 0;
1960 }
1961
1962 int Http2Upstream::redirect_to_https(Downstream *downstream) {
1963   auto &req = downstream->request();
1964   if (req.regular_connect_method() || req.scheme != "http") {
1965     return error_reply(downstream, 400);
1966   }
1967
1968   auto authority = util::extract_host(req.authority);
1969   if (authority.empty()) {
1970     return error_reply(downstream, 400);
1971   }
1972
1973   auto &balloc = downstream->get_block_allocator();
1974   auto config = get_config();
1975   auto &httpconf = config->http;
1976
1977   StringRef loc;
1978   if (httpconf.redirect_https_port == StringRef::from_lit("443")) {
1979     loc = concat_string_ref(balloc, StringRef::from_lit("https://"), authority,
1980                             req.path);
1981   } else {
1982     loc = concat_string_ref(balloc, StringRef::from_lit("https://"), authority,
1983                             StringRef::from_lit(":"),
1984                             httpconf.redirect_https_port, req.path);
1985   }
1986
1987   auto &resp = downstream->response();
1988   resp.http_status = 308;
1989   resp.fs.add_header_token(StringRef::from_lit("location"), loc, false,
1990                            http2::HD_LOCATION);
1991
1992   return send_reply(downstream, nullptr, 0);
1993 }
1994
1995 int Http2Upstream::consume(int32_t stream_id, size_t len) {
1996   int rv;
1997
1998   auto faddr = handler_->get_upstream_addr();
1999
2000   if (faddr->alt_mode != UpstreamAltMode::NONE) {
2001     return 0;
2002   }
2003
2004   rv = nghttp2_session_consume(session_, stream_id, len);
2005
2006   if (rv != 0) {
2007     ULOG(WARN, this) << "nghttp2_session_consume() returned error: "
2008                      << nghttp2_strerror(rv);
2009     return -1;
2010   }
2011
2012   return 0;
2013 }
2014
2015 void Http2Upstream::log_response_headers(
2016     Downstream *downstream, const std::vector<nghttp2_nv> &nva) const {
2017   std::stringstream ss;
2018   for (auto &nv : nva) {
2019     ss << TTY_HTTP_HD << StringRef{nv.name, nv.namelen} << TTY_RST << ": "
2020        << StringRef{nv.value, nv.valuelen} << "\n";
2021   }
2022   ULOG(INFO, this) << "HTTP response headers. stream_id="
2023                    << downstream->get_stream_id() << "\n"
2024                    << ss.str();
2025 }
2026
2027 int Http2Upstream::on_timeout(Downstream *downstream) {
2028   if (LOG_ENABLED(INFO)) {
2029     ULOG(INFO, this) << "Stream timeout stream_id="
2030                      << downstream->get_stream_id();
2031   }
2032
2033   rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
2034   handler_->signal_write();
2035
2036   return 0;
2037 }
2038
2039 void Http2Upstream::on_handler_delete() {
2040   for (auto d = downstream_queue_.get_downstreams(); d; d = d->dlnext) {
2041     if (d->get_dispatch_state() == DispatchState::ACTIVE &&
2042         d->accesslog_ready()) {
2043       handler_->write_accesslog(d);
2044     }
2045   }
2046 }
2047
2048 int Http2Upstream::on_downstream_reset(Downstream *downstream, bool no_retry) {
2049   int rv;
2050
2051   if (downstream->get_dispatch_state() != DispatchState::ACTIVE) {
2052     // This is error condition when we failed push_request_headers()
2053     // in initiate_downstream().  Otherwise, we have
2054     // DispatchState::ACTIVE state, or we did not set
2055     // DownstreamConnection.
2056     downstream->pop_downstream_connection();
2057     handler_->signal_write();
2058
2059     return 0;
2060   }
2061
2062   if (!downstream->request_submission_ready()) {
2063     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
2064       // We have got all response body already.  Send it off.
2065       downstream->pop_downstream_connection();
2066       return 0;
2067     }
2068     // pushed stream is handled here
2069     rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
2070     downstream->pop_downstream_connection();
2071
2072     handler_->signal_write();
2073
2074     return 0;
2075   }
2076
2077   downstream->pop_downstream_connection();
2078
2079   downstream->add_retry();
2080
2081   std::unique_ptr<DownstreamConnection> dconn;
2082
2083   rv = 0;
2084
2085   if (no_retry || downstream->no_more_retry()) {
2086     goto fail;
2087   }
2088
2089   // downstream connection is clean; we can retry with new
2090   // downstream connection.
2091
2092   for (;;) {
2093     auto dconn = handler_->get_downstream_connection(rv, downstream);
2094     if (!dconn) {
2095       goto fail;
2096     }
2097
2098     rv = downstream->attach_downstream_connection(std::move(dconn));
2099     if (rv == 0) {
2100       break;
2101     }
2102   }
2103
2104   rv = downstream->push_request_headers();
2105   if (rv != 0) {
2106     goto fail;
2107   }
2108
2109   return 0;
2110
2111 fail:
2112   if (rv == SHRPX_ERR_TLS_REQUIRED) {
2113     rv = on_downstream_abort_request_with_https_redirect(downstream);
2114   } else {
2115     rv = on_downstream_abort_request(downstream, 502);
2116   }
2117   if (rv != 0) {
2118     rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
2119   }
2120   downstream->pop_downstream_connection();
2121
2122   handler_->signal_write();
2123
2124   return 0;
2125 }
2126
2127 int Http2Upstream::prepare_push_promise(Downstream *downstream) {
2128   int rv;
2129
2130   const auto &req = downstream->request();
2131   auto &resp = downstream->response();
2132
2133   auto base = http2::get_pure_path_component(req.path);
2134   if (base.empty()) {
2135     return 0;
2136   }
2137
2138   auto &balloc = downstream->get_block_allocator();
2139
2140   for (auto &kv : resp.fs.headers()) {
2141     if (kv.token != http2::HD_LINK) {
2142       continue;
2143     }
2144     for (auto &link : http2::parse_link_header(kv.value)) {
2145       StringRef scheme, authority, path;
2146
2147       rv = http2::construct_push_component(balloc, scheme, authority, path,
2148                                            base, link.uri);
2149       if (rv != 0) {
2150         continue;
2151       }
2152
2153       if (scheme.empty()) {
2154         scheme = req.scheme;
2155       }
2156
2157       if (authority.empty()) {
2158         authority = req.authority;
2159       }
2160
2161       if (resp.is_resource_pushed(scheme, authority, path)) {
2162         continue;
2163       }
2164
2165       rv = submit_push_promise(scheme, authority, path, downstream);
2166       if (rv != 0) {
2167         return -1;
2168       }
2169
2170       resp.resource_pushed(scheme, authority, path);
2171     }
2172   }
2173   return 0;
2174 }
2175
2176 int Http2Upstream::submit_push_promise(const StringRef &scheme,
2177                                        const StringRef &authority,
2178                                        const StringRef &path,
2179                                        Downstream *downstream) {
2180   const auto &req = downstream->request();
2181
2182   std::vector<nghttp2_nv> nva;
2183   // 4 for :method, :scheme, :path and :authority
2184   nva.reserve(4 + req.fs.headers().size());
2185
2186   // juse use "GET" for now
2187   nva.push_back(http2::make_nv_ll(":method", "GET"));
2188   nva.push_back(http2::make_nv_ls_nocopy(":scheme", scheme));
2189   nva.push_back(http2::make_nv_ls_nocopy(":path", path));
2190   nva.push_back(http2::make_nv_ls_nocopy(":authority", authority));
2191
2192   for (auto &kv : req.fs.headers()) {
2193     switch (kv.token) {
2194     // TODO generate referer
2195     case http2::HD__AUTHORITY:
2196     case http2::HD__SCHEME:
2197     case http2::HD__METHOD:
2198     case http2::HD__PATH:
2199       continue;
2200     case http2::HD_ACCEPT_ENCODING:
2201     case http2::HD_ACCEPT_LANGUAGE:
2202     case http2::HD_CACHE_CONTROL:
2203     case http2::HD_HOST:
2204     case http2::HD_USER_AGENT:
2205       nva.push_back(http2::make_nv_nocopy(kv.name, kv.value, kv.no_index));
2206       break;
2207     }
2208   }
2209
2210   auto promised_stream_id = nghttp2_submit_push_promise(
2211       session_, NGHTTP2_FLAG_NONE, downstream->get_stream_id(), nva.data(),
2212       nva.size(), nullptr);
2213
2214   if (promised_stream_id < 0) {
2215     if (LOG_ENABLED(INFO)) {
2216       ULOG(INFO, this) << "nghttp2_submit_push_promise() failed: "
2217                        << nghttp2_strerror(promised_stream_id);
2218     }
2219     if (nghttp2_is_fatal(promised_stream_id)) {
2220       return -1;
2221     }
2222     return 0;
2223   }
2224
2225   if (LOG_ENABLED(INFO)) {
2226     std::stringstream ss;
2227     for (auto &nv : nva) {
2228       ss << TTY_HTTP_HD << StringRef{nv.name, nv.namelen} << TTY_RST << ": "
2229          << StringRef{nv.value, nv.valuelen} << "\n";
2230     }
2231     ULOG(INFO, this) << "HTTP push request headers. promised_stream_id="
2232                      << promised_stream_id << "\n"
2233                      << ss.str();
2234   }
2235
2236   return 0;
2237 }
2238
2239 bool Http2Upstream::push_enabled() const {
2240   auto config = get_config();
2241   return !(config->http2.no_server_push ||
2242            nghttp2_session_get_remote_settings(
2243                session_, NGHTTP2_SETTINGS_ENABLE_PUSH) == 0 ||
2244            config->http2_proxy);
2245 }
2246
2247 int Http2Upstream::initiate_push(Downstream *downstream, const StringRef &uri) {
2248   int rv;
2249
2250   if (uri.empty() || !push_enabled() ||
2251       (downstream->get_stream_id() % 2) == 0) {
2252     return 0;
2253   }
2254
2255   const auto &req = downstream->request();
2256
2257   auto base = http2::get_pure_path_component(req.path);
2258   if (base.empty()) {
2259     return -1;
2260   }
2261
2262   auto &balloc = downstream->get_block_allocator();
2263
2264   StringRef scheme, authority, path;
2265
2266   rv = http2::construct_push_component(balloc, scheme, authority, path, base,
2267                                        uri);
2268   if (rv != 0) {
2269     return -1;
2270   }
2271
2272   if (scheme.empty()) {
2273     scheme = req.scheme;
2274   }
2275
2276   if (authority.empty()) {
2277     authority = req.authority;
2278   }
2279
2280   auto &resp = downstream->response();
2281
2282   if (resp.is_resource_pushed(scheme, authority, path)) {
2283     return 0;
2284   }
2285
2286   rv = submit_push_promise(scheme, authority, path, downstream);
2287
2288   if (rv != 0) {
2289     return -1;
2290   }
2291
2292   resp.resource_pushed(scheme, authority, path);
2293
2294   return 0;
2295 }
2296
2297 int Http2Upstream::response_riovec(struct iovec *iov, int iovcnt) const {
2298   if (iovcnt == 0 || wb_.rleft() == 0) {
2299     return 0;
2300   }
2301
2302   return wb_.riovec(iov, iovcnt);
2303 }
2304
2305 void Http2Upstream::response_drain(size_t n) { wb_.drain(n); }
2306
2307 bool Http2Upstream::response_empty() const { return wb_.rleft() == 0; }
2308
2309 DefaultMemchunks *Http2Upstream::get_response_buf() { return &wb_; }
2310
2311 Downstream *
2312 Http2Upstream::on_downstream_push_promise(Downstream *downstream,
2313                                           int32_t promised_stream_id) {
2314   // promised_stream_id is for backend HTTP/2 session, not for
2315   // frontend.
2316   auto promised_downstream =
2317       std::make_unique<Downstream>(this, handler_->get_mcpool(), 0);
2318   auto &promised_req = promised_downstream->request();
2319
2320   promised_downstream->set_downstream_stream_id(promised_stream_id);
2321   // Set associated stream in frontend
2322   promised_downstream->set_assoc_stream_id(downstream->get_stream_id());
2323
2324   promised_downstream->disable_upstream_rtimer();
2325
2326   promised_req.http_major = 2;
2327   promised_req.http_minor = 0;
2328
2329   promised_req.fs.content_length = 0;
2330   promised_req.http2_expect_body = false;
2331
2332   auto ptr = promised_downstream.get();
2333   add_pending_downstream(std::move(promised_downstream));
2334   downstream_queue_.mark_active(ptr);
2335
2336   return ptr;
2337 }
2338
2339 int Http2Upstream::on_downstream_push_promise_complete(
2340     Downstream *downstream, Downstream *promised_downstream) {
2341   std::vector<nghttp2_nv> nva;
2342
2343   const auto &promised_req = promised_downstream->request();
2344   const auto &headers = promised_req.fs.headers();
2345
2346   nva.reserve(headers.size());
2347
2348   for (auto &kv : headers) {
2349     nva.push_back(http2::make_nv(kv.name, kv.value, kv.no_index));
2350   }
2351
2352   auto promised_stream_id = nghttp2_submit_push_promise(
2353       session_, NGHTTP2_FLAG_NONE, downstream->get_stream_id(), nva.data(),
2354       nva.size(), promised_downstream);
2355   if (promised_stream_id < 0) {
2356     return -1;
2357   }
2358
2359   promised_downstream->set_stream_id(promised_stream_id);
2360
2361   return 0;
2362 }
2363
2364 void Http2Upstream::cancel_premature_downstream(
2365     Downstream *promised_downstream) {
2366   if (LOG_ENABLED(INFO)) {
2367     ULOG(INFO, this) << "Remove premature promised stream "
2368                      << promised_downstream;
2369   }
2370   downstream_queue_.remove_and_get_blocked(promised_downstream, false);
2371 }
2372
2373 size_t Http2Upstream::get_max_buffer_size() const { return max_buffer_size_; }
2374
2375 } // namespace shrpx