src/shrpx_http3_upstream.cc

   1 /*
   2  * nghttp2 - HTTP/2 C Library
   3  *
   4  * Copyright (c) 2021 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_http3_upstream.h"
  26
  27 #include <sys/types.h>
  28 #include <sys/stat.h>
  29 #include <fcntl.h>
  30 #include <netinet/udp.h>
  31
  32 #include <cstdio>
  33
  34 #include <ngtcp2/ngtcp2_crypto.h>
  35
  36 #include "shrpx_client_handler.h"
  37 #include "shrpx_downstream.h"
  38 #include "shrpx_downstream_connection.h"
  39 #include "shrpx_log.h"
  40 #include "shrpx_quic.h"
  41 #include "shrpx_worker.h"
  42 #include "shrpx_http.h"
  43 #include "shrpx_connection_handler.h"
  44 #ifdef HAVE_MRUBY
  45 #  include "shrpx_mruby.h"
  46 #endif // HAVE_MRUBY
  47 #include "http3.h"
  48 #include "util.h"
  49
  50 namespace shrpx {
  51
  52 namespace {
  53 void idle_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
  54   auto upstream = static_cast<Http3Upstream *>(w->data);
  55
  56   if (LOG_ENABLED(INFO)) {
  57     ULOG(INFO, upstream) << "QUIC idle timeout";
  58   }
  59
  60   upstream->idle_close();
  61
  62   auto handler = upstream->get_client_handler();
  63
  64   delete handler;
  65 }
  66 } // namespace
  67
  68 namespace {
  69 void timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
  70   auto upstream = static_cast<Http3Upstream *>(w->data);
  71
  72   if (upstream->handle_expiry() != 0 || upstream->on_write() != 0) {
  73     goto fail;
  74   }
  75
  76   return;
  77
  78 fail:
  79   auto handler = upstream->get_client_handler();
  80
  81   delete handler;
  82 }
  83 } // namespace
  84
  85 namespace {
  86 void shutdown_timeout_cb(struct ev_loop *loop, ev_timer *w, int revents) {
  87   auto upstream = static_cast<Http3Upstream *>(w->data);
  88   auto handler = upstream->get_client_handler();
  89
  90   if (upstream->submit_goaway() != 0) {
  91     delete handler;
  92   }
  93 }
  94 } // namespace
  95
  96 namespace {
  97 void prepare_cb(struct ev_loop *loop, ev_prepare *w, int revent) {
  98   auto upstream = static_cast<Http3Upstream *>(w->data);
  99   auto handler = upstream->get_client_handler();
 100
 101   if (upstream->check_shutdown() != 0) {
 102     delete handler;
 103   }
 104 }
 105 } // namespace
 106
 107 namespace {
 108 size_t downstream_queue_size(Worker *worker) {
 109   auto &downstreamconf = *worker->get_downstream_config();
 110
 111   if (get_config()->http2_proxy) {
 112     return downstreamconf.connections_per_host;
 113   }
 114
 115   return downstreamconf.connections_per_frontend;
 116 }
 117 } // namespace
 118
 119 Http3Upstream::Http3Upstream(ClientHandler *handler)
 120     : handler_{handler},
 121       max_udp_payload_size_{SHRPX_QUIC_MAX_UDP_PAYLOAD_SIZE},
 122       qlog_fd_{-1},
 123       hashed_scid_{},
 124       conn_{nullptr},
 125       tls_alert_{0},
 126       httpconn_{nullptr},
 127       downstream_queue_{downstream_queue_size(handler->get_worker()),
 128                         !get_config()->http2_proxy},
 129       idle_close_{false},
 130       retry_close_{false} {
 131   ev_timer_init(&timer_, timeoutcb, 0., 0.);
 132   timer_.data = this;
 133
 134   auto config = get_config();
 135   auto &quicconf = config->quic;
 136
 137   ev_timer_init(&idle_timer_, idle_timeoutcb, 0.,
 138                 quicconf.upstream.timeout.idle);
 139   idle_timer_.data = this;
 140
 141   ev_timer_init(&shutdown_timer_, shutdown_timeout_cb, 0., 0.);
 142   shutdown_timer_.data = this;
 143
 144   ev_prepare_init(&prep_, prepare_cb);
 145   prep_.data = this;
 146   ev_prepare_start(handler_->get_loop(), &prep_);
 147 }
 148
 149 Http3Upstream::~Http3Upstream() {
 150   auto loop = handler_->get_loop();
 151
 152   ev_prepare_stop(loop, &prep_);
 153   ev_timer_stop(loop, &shutdown_timer_);
 154   ev_timer_stop(loop, &idle_timer_);
 155   ev_timer_stop(loop, &timer_);
 156
 157   nghttp3_conn_del(httpconn_);
 158
 159   ngtcp2_conn_del(conn_);
 160
 161   if (qlog_fd_ != -1) {
 162     close(qlog_fd_);
 163   }
 164 }
 165
 166 namespace {
 167 void log_printf(void *user_data, const char *fmt, ...) {
 168   va_list ap;
 169   std::array<char, 4096> buf;
 170
 171   va_start(ap, fmt);
 172   auto nwrite = vsnprintf(buf.data(), buf.size(), fmt, ap);
 173   va_end(ap);
 174
 175   if (static_cast<size_t>(nwrite) >= buf.size()) {
 176     nwrite = buf.size() - 1;
 177   }
 178
 179   buf[nwrite++] = '\n';
 180
 181   while (write(fileno(stderr), buf.data(), nwrite) == -1 && errno == EINTR)
 182     ;
 183 }
 184 } // namespace
 185
 186 namespace {
 187 void qlog_write(void *user_data, uint32_t flags, const void *data,
 188                 size_t datalen) {
 189   auto upstream = static_cast<Http3Upstream *>(user_data);
 190
 191   upstream->qlog_write(data, datalen, flags & NGTCP2_QLOG_WRITE_FLAG_FIN);
 192 }
 193 } // namespace
 194
 195 void Http3Upstream::qlog_write(const void *data, size_t datalen, bool fin) {
 196   assert(qlog_fd_ != -1);
 197
 198   ssize_t nwrite;
 199
 200   while ((nwrite = write(qlog_fd_, data, datalen)) == -1 && errno == EINTR)
 201     ;
 202
 203   if (fin) {
 204     close(qlog_fd_);
 205     qlog_fd_ = -1;
 206   }
 207 }
 208
 209 namespace {
 210 void rand(uint8_t *dest, size_t destlen, const ngtcp2_rand_ctx *rand_ctx) {
 211   util::random_bytes(dest, dest + destlen,
 212                      *static_cast<std::mt19937 *>(rand_ctx->native_handle));
 213 }
 214 } // namespace
 215
 216 namespace {
 217 int get_new_connection_id(ngtcp2_conn *conn, ngtcp2_cid *cid, uint8_t *token,
 218                           size_t cidlen, void *user_data) {
 219   auto upstream = static_cast<Http3Upstream *>(user_data);
 220   auto handler = upstream->get_client_handler();
 221   auto worker = handler->get_worker();
 222   auto conn_handler = worker->get_connection_handler();
 223   auto &qkms = conn_handler->get_quic_keying_materials();
 224   auto &qkm = qkms->keying_materials.front();
 225
 226   if (generate_quic_connection_id(*cid, cidlen, worker->get_cid_prefix(),
 227                                   qkm.id, qkm.cid_encryption_key.data()) != 0) {
 228     return NGTCP2_ERR_CALLBACK_FAILURE;
 229   }
 230
 231   if (generate_quic_stateless_reset_token(token, *cid, qkm.secret.data(),
 232                                           qkm.secret.size()) != 0) {
 233     return NGTCP2_ERR_CALLBACK_FAILURE;
 234   }
 235
 236   auto quic_connection_handler = worker->get_quic_connection_handler();
 237
 238   quic_connection_handler->add_connection_id(*cid, handler);
 239
 240   return 0;
 241 }
 242 } // namespace
 243
 244 namespace {
 245 int remove_connection_id(ngtcp2_conn *conn, const ngtcp2_cid *cid,
 246                          void *user_data) {
 247   auto upstream = static_cast<Http3Upstream *>(user_data);
 248   auto handler = upstream->get_client_handler();
 249   auto worker = handler->get_worker();
 250   auto quic_conn_handler = worker->get_quic_connection_handler();
 251
 252   quic_conn_handler->remove_connection_id(*cid);
 253
 254   return 0;
 255 }
 256 } // namespace
 257
 258 void Http3Upstream::http_begin_request_headers(int64_t stream_id) {
 259   auto downstream =
 260       std::make_unique<Downstream>(this, handler_->get_mcpool(), stream_id);
 261   nghttp3_conn_set_stream_user_data(httpconn_, stream_id, downstream.get());
 262
 263   downstream->reset_upstream_rtimer();
 264
 265   handler_->repeat_read_timer();
 266
 267   auto &req = downstream->request();
 268   req.http_major = 3;
 269   req.http_minor = 0;
 270
 271   add_pending_downstream(std::move(downstream));
 272 }
 273
 274 void Http3Upstream::add_pending_downstream(
 275     std::unique_ptr<Downstream> downstream) {
 276   downstream_queue_.add_pending(std::move(downstream));
 277 }
 278
 279 namespace {
 280 int recv_stream_data(ngtcp2_conn *conn, uint32_t flags, int64_t stream_id,
 281                      uint64_t offset, const uint8_t *data, size_t datalen,
 282                      void *user_data, void *stream_user_data) {
 283   auto upstream = static_cast<Http3Upstream *>(user_data);
 284
 285   if (upstream->recv_stream_data(flags, stream_id, data, datalen) != 0) {
 286     return NGTCP2_ERR_CALLBACK_FAILURE;
 287   }
 288
 289   return 0;
 290 }
 291 } // namespace
 292
 293 int Http3Upstream::recv_stream_data(uint32_t flags, int64_t stream_id,
 294                                     const uint8_t *data, size_t datalen) {
 295   assert(httpconn_);
 296
 297   auto nconsumed = nghttp3_conn_read_stream(
 298       httpconn_, stream_id, data, datalen, flags & NGTCP2_STREAM_DATA_FLAG_FIN);
 299   if (nconsumed < 0) {
 300     ULOG(ERROR, this) << "nghttp3_conn_read_stream: "
 301                       << nghttp3_strerror(nconsumed);
 302     last_error_ = quic::err_application(nconsumed);
 303     return -1;
 304   }
 305
 306   ngtcp2_conn_extend_max_stream_offset(conn_, stream_id, nconsumed);
 307   ngtcp2_conn_extend_max_offset(conn_, nconsumed);
 308
 309   return 0;
 310 }
 311
 312 namespace {
 313 int stream_close(ngtcp2_conn *conn, uint32_t flags, int64_t stream_id,
 314                  uint64_t app_error_code, void *user_data,
 315                  void *stream_user_data) {
 316   auto upstream = static_cast<Http3Upstream *>(user_data);
 317
 318   if (!(flags & NGTCP2_STREAM_CLOSE_FLAG_APP_ERROR_CODE_SET)) {
 319     app_error_code = NGHTTP3_H3_NO_ERROR;
 320   }
 321
 322   if (upstream->stream_close(stream_id, app_error_code) != 0) {
 323     return NGTCP2_ERR_CALLBACK_FAILURE;
 324   }
 325
 326   return 0;
 327 }
 328 } // namespace
 329
 330 int Http3Upstream::stream_close(int64_t stream_id, uint64_t app_error_code) {
 331   if (!httpconn_) {
 332     return 0;
 333   }
 334
 335   auto rv = nghttp3_conn_close_stream(httpconn_, stream_id, app_error_code);
 336   switch (rv) {
 337   case 0:
 338     break;
 339   case NGHTTP3_ERR_STREAM_NOT_FOUND:
 340     if (ngtcp2_is_bidi_stream(stream_id)) {
 341       ngtcp2_conn_extend_max_streams_bidi(conn_, 1);
 342     }
 343     break;
 344   default:
 345     ULOG(ERROR, this) << "nghttp3_conn_close_stream: " << nghttp3_strerror(rv);
 346     last_error_ = quic::err_application(rv);
 347     return -1;
 348   }
 349
 350   return 0;
 351 }
 352
 353 namespace {
 354 int acked_stream_data_offset(ngtcp2_conn *conn, int64_t stream_id,
 355                              uint64_t offset, uint64_t datalen, void *user_data,
 356                              void *stream_user_data) {
 357   auto upstream = static_cast<Http3Upstream *>(user_data);
 358
 359   if (upstream->acked_stream_data_offset(stream_id, datalen) != 0) {
 360     return NGTCP2_ERR_CALLBACK_FAILURE;
 361   }
 362
 363   return 0;
 364 }
 365 } // namespace
 366
 367 int Http3Upstream::acked_stream_data_offset(int64_t stream_id,
 368                                             uint64_t datalen) {
 369   if (!httpconn_) {
 370     return 0;
 371   }
 372
 373   auto rv = nghttp3_conn_add_ack_offset(httpconn_, stream_id, datalen);
 374   if (rv != 0) {
 375     ULOG(ERROR, this) << "nghttp3_conn_add_ack_offset: "
 376                       << nghttp3_strerror(rv);
 377     return -1;
 378   }
 379
 380   return 0;
 381 }
 382
 383 namespace {
 384 int extend_max_stream_data(ngtcp2_conn *conn, int64_t stream_id,
 385                            uint64_t max_data, void *user_data,
 386                            void *stream_user_data) {
 387   auto upstream = static_cast<Http3Upstream *>(user_data);
 388
 389   if (upstream->extend_max_stream_data(stream_id) != 0) {
 390     return NGTCP2_ERR_CALLBACK_FAILURE;
 391   }
 392
 393   return 0;
 394 }
 395 } // namespace
 396
 397 int Http3Upstream::extend_max_stream_data(int64_t stream_id) {
 398   if (!httpconn_) {
 399     return 0;
 400   }
 401
 402   auto rv = nghttp3_conn_unblock_stream(httpconn_, stream_id);
 403   if (rv != 0) {
 404     ULOG(ERROR, this) << "nghttp3_conn_unblock_stream: "
 405                       << nghttp3_strerror(rv);
 406     return -1;
 407   }
 408
 409   return 0;
 410 }
 411
 412 namespace {
 413 int extend_max_remote_streams_bidi(ngtcp2_conn *conn, uint64_t max_streams,
 414                                    void *user_data) {
 415   auto upstream = static_cast<Http3Upstream *>(user_data);
 416
 417   upstream->extend_max_remote_streams_bidi(max_streams);
 418
 419   return 0;
 420 }
 421 } // namespace
 422
 423 void Http3Upstream::extend_max_remote_streams_bidi(uint64_t max_streams) {
 424   nghttp3_conn_set_max_client_streams_bidi(httpconn_, max_streams);
 425 }
 426
 427 namespace {
 428 int stream_reset(ngtcp2_conn *conn, int64_t stream_id, uint64_t final_size,
 429                  uint64_t app_error_code, void *user_data,
 430                  void *stream_user_data) {
 431   auto upstream = static_cast<Http3Upstream *>(user_data);
 432
 433   if (upstream->http_shutdown_stream_read(stream_id) != 0) {
 434     return NGTCP2_ERR_CALLBACK_FAILURE;
 435   }
 436
 437   return 0;
 438 }
 439 } // namespace
 440
 441 int Http3Upstream::http_shutdown_stream_read(int64_t stream_id) {
 442   if (!httpconn_) {
 443     return 0;
 444   }
 445
 446   auto rv = nghttp3_conn_shutdown_stream_read(httpconn_, stream_id);
 447   if (rv != 0) {
 448     ULOG(ERROR, this) << "nghttp3_conn_shutdown_stream_read: "
 449                       << nghttp3_strerror(rv);
 450     return -1;
 451   }
 452
 453   return 0;
 454 }
 455
 456 namespace {
 457 int stream_stop_sending(ngtcp2_conn *conn, int64_t stream_id,
 458                         uint64_t app_error_code, void *user_data,
 459                         void *stream_user_data) {
 460   auto upstream = static_cast<Http3Upstream *>(user_data);
 461
 462   if (upstream->http_shutdown_stream_read(stream_id) != 0) {
 463     return NGTCP2_ERR_CALLBACK_FAILURE;
 464   }
 465
 466   return 0;
 467 }
 468 } // namespace
 469
 470 namespace {
 471 int handshake_completed(ngtcp2_conn *conn, void *user_data) {
 472   auto upstream = static_cast<Http3Upstream *>(user_data);
 473
 474   if (upstream->handshake_completed() != 0) {
 475     return NGTCP2_ERR_CALLBACK_FAILURE;
 476   }
 477
 478   return 0;
 479 }
 480 } // namespace
 481
 482 int Http3Upstream::handshake_completed() {
 483   handler_->set_alpn_from_conn();
 484
 485   auto alpn = handler_->get_alpn();
 486   if (alpn.empty()) {
 487     ULOG(ERROR, this) << "NO ALPN was negotiated";
 488     return -1;
 489   }
 490
 491   std::array<uint8_t, NGTCP2_CRYPTO_MAX_REGULAR_TOKENLEN> token;
 492   size_t tokenlen;
 493
 494   auto path = ngtcp2_conn_get_path(conn_);
 495   auto worker = handler_->get_worker();
 496   auto conn_handler = worker->get_connection_handler();
 497   auto &qkms = conn_handler->get_quic_keying_materials();
 498   auto &qkm = qkms->keying_materials.front();
 499
 500   if (generate_token(token.data(), tokenlen, path->remote.addr,
 501                      path->remote.addrlen, qkm.secret.data(),
 502                      qkm.secret.size()) != 0) {
 503     return 0;
 504   }
 505
 506   auto rv = ngtcp2_conn_submit_new_token(conn_, token.data(), tokenlen);
 507   if (rv != 0) {
 508     ULOG(ERROR, this) << "ngtcp2_conn_submit_new_token: "
 509                       << ngtcp2_strerror(rv);
 510     return -1;
 511   }
 512
 513   return 0;
 514 }
 515
 516 int Http3Upstream::init(const UpstreamAddr *faddr, const Address &remote_addr,
 517                         const Address &local_addr,
 518                         const ngtcp2_pkt_hd &initial_hd,
 519                         const ngtcp2_cid *odcid, const uint8_t *token,
 520                         size_t tokenlen) {
 521   int rv;
 522
 523   auto worker = handler_->get_worker();
 524   auto conn_handler = worker->get_connection_handler();
 525
 526   auto callbacks = ngtcp2_callbacks{
 527       nullptr, // client_initial
 528       ngtcp2_crypto_recv_client_initial_cb,
 529       ngtcp2_crypto_recv_crypto_data_cb,
 530       shrpx::handshake_completed,
 531       nullptr, // recv_version_negotiation
 532       ngtcp2_crypto_encrypt_cb,
 533       ngtcp2_crypto_decrypt_cb,
 534       ngtcp2_crypto_hp_mask_cb,
 535       shrpx::recv_stream_data,
 536       shrpx::acked_stream_data_offset,
 537       nullptr, // stream_open
 538       shrpx::stream_close,
 539       nullptr, // recv_stateless_reset
 540       nullptr, // recv_retry
 541       nullptr, // extend_max_local_streams_bidi
 542       nullptr, // extend_max_local_streams_uni
 543       rand,
 544       get_new_connection_id,
 545       remove_connection_id,
 546       ngtcp2_crypto_update_key_cb,
 547       nullptr, // path_validation
 548       nullptr, // select_preferred_addr
 549       shrpx::stream_reset,
 550       shrpx::extend_max_remote_streams_bidi,
 551       nullptr, // extend_max_remote_streams_uni
 552       shrpx::extend_max_stream_data,
 553       nullptr, // dcid_status
 554       nullptr, // handshake_confirmed
 555       nullptr, // recv_new_token
 556       ngtcp2_crypto_delete_crypto_aead_ctx_cb,
 557       ngtcp2_crypto_delete_crypto_cipher_ctx_cb,
 558       nullptr, // recv_datagram
 559       nullptr, // ack_datagram
 560       nullptr, // lost_datagram
 561       ngtcp2_crypto_get_path_challenge_data_cb,
 562       shrpx::stream_stop_sending,
 563   };
 564
 565   auto config = get_config();
 566   auto &quicconf = config->quic;
 567   auto &http3conf = config->http3;
 568
 569   auto &qkms = conn_handler->get_quic_keying_materials();
 570   auto &qkm = qkms->keying_materials.front();
 571
 572   ngtcp2_cid scid;
 573
 574   if (generate_quic_connection_id(scid, SHRPX_QUIC_SCIDLEN,
 575                                   worker->get_cid_prefix(), qkm.id,
 576                                   qkm.cid_encryption_key.data()) != 0) {
 577     return -1;
 578   }
 579
 580   ngtcp2_settings settings;
 581   ngtcp2_settings_default(&settings);
 582   if (quicconf.upstream.debug.log) {
 583     settings.log_printf = log_printf;
 584   }
 585
 586   if (!quicconf.upstream.qlog.dir.empty()) {
 587     auto fd = open_qlog_file(quicconf.upstream.qlog.dir, scid);
 588     if (fd != -1) {
 589       qlog_fd_ = fd;
 590       settings.qlog.odcid = initial_hd.dcid;
 591       settings.qlog.write = shrpx::qlog_write;
 592     }
 593   }
 594
 595   settings.initial_ts = quic_timestamp();
 596   settings.initial_rtt = static_cast<ngtcp2_tstamp>(
 597       quicconf.upstream.initial_rtt * NGTCP2_SECONDS);
 598   settings.cc_algo = quicconf.upstream.congestion_controller;
 599   settings.max_window = http3conf.upstream.max_connection_window_size;
 600   settings.max_stream_window = http3conf.upstream.max_window_size;
 601   settings.max_udp_payload_size = SHRPX_QUIC_MAX_UDP_PAYLOAD_SIZE;
 602   settings.assume_symmetric_path = 1;
 603   settings.rand_ctx.native_handle = &worker->get_randgen();
 604   settings.token = ngtcp2_vec{const_cast<uint8_t *>(token), tokenlen};
 605
 606   ngtcp2_transport_params params;
 607   ngtcp2_transport_params_default(&params);
 608   params.initial_max_streams_bidi = http3conf.upstream.max_concurrent_streams;
 609   // The minimum number of unidirectional streams required for HTTP/3.
 610   params.initial_max_streams_uni = 3;
 611   params.initial_max_data = http3conf.upstream.connection_window_size;
 612   params.initial_max_stream_data_bidi_remote = http3conf.upstream.window_size;
 613   params.initial_max_stream_data_uni = http3conf.upstream.window_size;
 614   params.max_idle_timeout = static_cast<ngtcp2_tstamp>(
 615       quicconf.upstream.timeout.idle * NGTCP2_SECONDS);
 616
 617 #ifdef OPENSSL_IS_BORINGSSL
 618   if (quicconf.upstream.early_data) {
 619     ngtcp2_transport_params early_data_params{
 620         .initial_max_stream_data_bidi_local =
 621             params.initial_max_stream_data_bidi_local,
 622         .initial_max_stream_data_bidi_remote =
 623             params.initial_max_stream_data_bidi_remote,
 624         .initial_max_stream_data_uni = params.initial_max_stream_data_uni,
 625         .initial_max_data = params.initial_max_data,
 626         .initial_max_streams_bidi = params.initial_max_streams_bidi,
 627         .initial_max_streams_uni = params.initial_max_streams_uni,
 628     };
 629
 630     // TODO include HTTP/3 SETTINGS
 631
 632     std::array<uint8_t, 128> quic_early_data_ctx;
 633
 634     auto quic_early_data_ctxlen = ngtcp2_encode_transport_params(
 635         quic_early_data_ctx.data(), quic_early_data_ctx.size(),
 636         NGTCP2_TRANSPORT_PARAMS_TYPE_ENCRYPTED_EXTENSIONS, &early_data_params);
 637
 638     assert(quic_early_data_ctxlen > 0);
 639     assert(static_cast<size_t>(quic_early_data_ctxlen) <=
 640            quic_early_data_ctx.size());
 641
 642     if (SSL_set_quic_early_data_context(handler_->get_ssl(),
 643                                         quic_early_data_ctx.data(),
 644                                         quic_early_data_ctxlen) != 1) {
 645       ULOG(ERROR, this) << "SSL_set_quic_early_data_context failed";
 646       return -1;
 647     }
 648   }
 649 #endif // OPENSSL_IS_BORINGSSL
 650
 651   if (odcid) {
 652     params.original_dcid = *odcid;
 653     params.retry_scid = initial_hd.dcid;
 654     params.retry_scid_present = 1;
 655   } else {
 656     params.original_dcid = initial_hd.dcid;
 657   }
 658
 659   rv = generate_quic_stateless_reset_token(
 660       params.stateless_reset_token, scid, qkm.secret.data(), qkm.secret.size());
 661   if (rv != 0) {
 662     ULOG(ERROR, this) << "generate_quic_stateless_reset_token failed";
 663     return -1;
 664   }
 665   params.stateless_reset_token_present = 1;
 666
 667   auto path = ngtcp2_path{
 668       {local_addr.len, const_cast<sockaddr *>(&local_addr.su.sa)},
 669       {remote_addr.len, const_cast<sockaddr *>(&remote_addr.su.sa)},
 670       const_cast<UpstreamAddr *>(faddr),
 671   };
 672
 673   rv = ngtcp2_conn_server_new(&conn_, &initial_hd.scid, &scid, &path,
 674                               initial_hd.version, &callbacks, &settings,
 675                               &params, nullptr, this);
 676   if (rv != 0) {
 677     ULOG(ERROR, this) << "ngtcp2_conn_server_new: " << ngtcp2_strerror(rv);
 678     return -1;
 679   }
 680
 681   ngtcp2_conn_set_tls_native_handle(conn_, handler_->get_ssl());
 682
 683   auto quic_connection_handler = worker->get_quic_connection_handler();
 684
 685   if (generate_quic_hashed_connection_id(hashed_scid_, remote_addr, local_addr,
 686                                          initial_hd.dcid) != 0) {
 687     return -1;
 688   }
 689
 690   quic_connection_handler->add_connection_id(hashed_scid_, handler_);
 691   quic_connection_handler->add_connection_id(scid, handler_);
 692
 693   return 0;
 694 }
 695
 696 int Http3Upstream::on_read() { return 0; }
 697
 698 int Http3Upstream::on_write() {
 699   if (write_streams() != 0) {
 700     return -1;
 701   }
 702
 703   reset_timer();
 704
 705   return 0;
 706 }
 707
 708 int Http3Upstream::write_streams() {
 709   std::array<nghttp3_vec, 16> vec;
 710   std::array<uint8_t, 64_k> buf;
 711   auto max_udp_payload_size = std::min(
 712       max_udp_payload_size_, ngtcp2_conn_get_path_max_udp_payload_size(conn_));
 713   size_t max_pktcnt =
 714       std::min(static_cast<size_t>(64_k), ngtcp2_conn_get_send_quantum(conn_)) /
 715       max_udp_payload_size;
 716   ngtcp2_pkt_info pi;
 717   uint8_t *bufpos = buf.data();
 718   ngtcp2_path_storage ps, prev_ps;
 719   size_t pktcnt = 0;
 720   int rv;
 721   auto ts = quic_timestamp();
 722
 723   ngtcp2_path_storage_zero(&ps);
 724   ngtcp2_path_storage_zero(&prev_ps);
 725
 726   auto config = get_config();
 727   auto &quicconf = config->quic;
 728
 729   if (quicconf.upstream.congestion_controller != NGTCP2_CC_ALGO_BBR) {
 730     max_pktcnt = std::min(max_pktcnt, static_cast<size_t>(10));
 731   }
 732
 733   for (;;) {
 734     int64_t stream_id = -1;
 735     int fin = 0;
 736     nghttp3_ssize sveccnt = 0;
 737
 738     if (httpconn_ && ngtcp2_conn_get_max_data_left(conn_)) {
 739       sveccnt = nghttp3_conn_writev_stream(httpconn_, &stream_id, &fin,
 740                                            vec.data(), vec.size());
 741       if (sveccnt < 0) {
 742         ULOG(ERROR, this) << "nghttp3_conn_writev_stream: "
 743                           << nghttp3_strerror(sveccnt);
 744         last_error_ = quic::err_application(sveccnt);
 745         return handle_error();
 746       }
 747     }
 748
 749     ngtcp2_ssize ndatalen;
 750     auto v = vec.data();
 751     auto vcnt = static_cast<size_t>(sveccnt);
 752
 753     uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_MORE;
 754     if (fin) {
 755       flags |= NGTCP2_WRITE_STREAM_FLAG_FIN;
 756     }
 757
 758     auto nwrite = ngtcp2_conn_writev_stream(
 759         conn_, &ps.path, &pi, bufpos, max_udp_payload_size, &ndatalen, flags,
 760         stream_id, reinterpret_cast<const ngtcp2_vec *>(v), vcnt, ts);
 761     if (nwrite < 0) {
 762       switch (nwrite) {
 763       case NGTCP2_ERR_STREAM_DATA_BLOCKED:
 764         assert(ndatalen == -1);
 765         rv = nghttp3_conn_block_stream(httpconn_, stream_id);
 766         if (rv != 0) {
 767           ULOG(ERROR, this)
 768               << "nghttp3_conn_block_stream: " << nghttp3_strerror(rv);
 769           last_error_ = quic::err_application(rv);
 770           return handle_error();
 771         }
 772         continue;
 773       case NGTCP2_ERR_STREAM_SHUT_WR:
 774         assert(ndatalen == -1);
 775         rv = nghttp3_conn_shutdown_stream_write(httpconn_, stream_id);
 776         if (rv != 0) {
 777           ULOG(ERROR, this)
 778               << "nghttp3_conn_shutdown_stream_write: " << nghttp3_strerror(rv);
 779           last_error_ = quic::err_application(rv);
 780           return handle_error();
 781         }
 782         continue;
 783       case NGTCP2_ERR_WRITE_MORE:
 784         assert(ndatalen >= 0);
 785         rv = nghttp3_conn_add_write_offset(httpconn_, stream_id, ndatalen);
 786         if (rv != 0) {
 787           ULOG(ERROR, this)
 788               << "nghttp3_conn_add_write_offset: " << nghttp3_strerror(rv);
 789           last_error_ = quic::err_application(rv);
 790           return handle_error();
 791         }
 792         continue;
 793       }
 794
 795       assert(ndatalen == -1);
 796
 797       ULOG(ERROR, this) << "ngtcp2_conn_writev_stream: "
 798                         << ngtcp2_strerror(nwrite);
 799
 800       last_error_ = quic::err_transport(nwrite);
 801
 802       handler_->get_connection()->wlimit.stopw();
 803
 804       return handle_error();
 805     } else if (ndatalen >= 0) {
 806       rv = nghttp3_conn_add_write_offset(httpconn_, stream_id, ndatalen);
 807       if (rv != 0) {
 808         ULOG(ERROR, this) << "nghttp3_conn_add_write_offset: "
 809                           << nghttp3_strerror(rv);
 810         last_error_ = quic::err_application(rv);
 811         return handle_error();
 812       }
 813     }
 814
 815     if (nwrite == 0) {
 816       if (bufpos - buf.data()) {
 817         send_packet(static_cast<UpstreamAddr *>(prev_ps.path.user_data),
 818                     prev_ps.path.remote.addr, prev_ps.path.remote.addrlen,
 819                     prev_ps.path.local.addr, prev_ps.path.local.addrlen,
 820                     buf.data(), bufpos - buf.data(), max_udp_payload_size);
 821
 822         reset_idle_timer();
 823       }
 824
 825       ngtcp2_conn_update_pkt_tx_time(conn_, ts);
 826
 827       handler_->get_connection()->wlimit.stopw();
 828
 829       return 0;
 830     }
 831
 832     bufpos += nwrite;
 833
 834 #ifdef UDP_SEGMENT
 835     if (pktcnt == 0) {
 836       ngtcp2_path_copy(&prev_ps.path, &ps.path);
 837     } else if (!ngtcp2_path_eq(&prev_ps.path, &ps.path)) {
 838       send_packet(static_cast<UpstreamAddr *>(prev_ps.path.user_data),
 839                   prev_ps.path.remote.addr, prev_ps.path.remote.addrlen,
 840                   prev_ps.path.local.addr, prev_ps.path.local.addrlen,
 841                   buf.data(), bufpos - buf.data() - nwrite,
 842                   max_udp_payload_size);
 843
 844       send_packet(static_cast<UpstreamAddr *>(ps.path.user_data),
 845                   ps.path.remote.addr, ps.path.remote.addrlen,
 846                   ps.path.local.addr, ps.path.local.addrlen, bufpos - nwrite,
 847                   nwrite, max_udp_payload_size);
 848
 849       ngtcp2_conn_update_pkt_tx_time(conn_, ts);
 850       reset_idle_timer();
 851
 852       handler_->signal_write();
 853
 854       return 0;
 855     }
 856
 857     if (++pktcnt == max_pktcnt ||
 858         static_cast<size_t>(nwrite) < max_udp_payload_size) {
 859       send_packet(static_cast<UpstreamAddr *>(ps.path.user_data),
 860                   ps.path.remote.addr, ps.path.remote.addrlen,
 861                   ps.path.local.addr, ps.path.local.addrlen, buf.data(),
 862                   bufpos - buf.data(), max_udp_payload_size);
 863
 864       ngtcp2_conn_update_pkt_tx_time(conn_, ts);
 865       reset_idle_timer();
 866
 867       handler_->signal_write();
 868
 869       return 0;
 870     }
 871 #else  // !UDP_SEGMENT
 872     send_packet(static_cast<UpstreamAddr *>(ps.path.user_data),
 873                 ps.path.remote.addr, ps.path.remote.addrlen, ps.path.local.addr,
 874                 ps.path.local.addrlen, buf.data(), bufpos - buf.data(), 0);
 875
 876     if (++pktcnt == max_pktcnt) {
 877       ngtcp2_conn_update_pkt_tx_time(conn_, ts);
 878       reset_idle_timer();
 879
 880       handler_->signal_write();
 881
 882       return 0;
 883     }
 884
 885     bufpos = buf.data();
 886 #endif // !UDP_SEGMENT
 887   }
 888
 889   return 0;
 890 }
 891
 892 int Http3Upstream::on_timeout(Downstream *downstream) { return 0; }
 893
 894 int Http3Upstream::on_downstream_abort_request(Downstream *downstream,
 895                                                unsigned int status_code) {
 896   int rv;
 897
 898   rv = error_reply(downstream, status_code);
 899
 900   if (rv != 0) {
 901     return -1;
 902   }
 903
 904   handler_->signal_write();
 905
 906   return 0;
 907 }
 908
 909 int Http3Upstream::on_downstream_abort_request_with_https_redirect(
 910     Downstream *downstream) {
 911   int rv;
 912
 913   rv = redirect_to_https(downstream);
 914   if (rv != 0) {
 915     return -1;
 916   }
 917
 918   handler_->signal_write();
 919   return 0;
 920 }
 921
 922 namespace {
 923 uint64_t
 924 infer_upstream_shutdown_stream_error_code(uint32_t downstream_error_code) {
 925   // NGHTTP2_REFUSED_STREAM is important because it tells upstream
 926   // client to retry.
 927   switch (downstream_error_code) {
 928   case NGHTTP2_NO_ERROR:
 929     return NGHTTP3_H3_NO_ERROR;
 930   case NGHTTP2_REFUSED_STREAM:
 931     return NGHTTP3_H3_REQUEST_REJECTED;
 932   default:
 933     return NGHTTP3_H3_INTERNAL_ERROR;
 934   }
 935 }
 936 } // namespace
 937
 938 int Http3Upstream::downstream_read(DownstreamConnection *dconn) {
 939   auto downstream = dconn->get_downstream();
 940
 941   if (downstream->get_response_state() == DownstreamState::MSG_RESET) {
 942     // The downstream stream was reset (canceled). In this case,
 943     // RST_STREAM to the upstream and delete downstream connection
 944     // here. Deleting downstream will be taken place at
 945     // on_stream_close_callback.
 946     shutdown_stream(downstream,
 947                     infer_upstream_shutdown_stream_error_code(
 948                         downstream->get_response_rst_stream_error_code()));
 949     downstream->pop_downstream_connection();
 950     // dconn was deleted
 951     dconn = nullptr;
 952   } else if (downstream->get_response_state() ==
 953              DownstreamState::MSG_BAD_HEADER) {
 954     if (error_reply(downstream, 502) != 0) {
 955       return -1;
 956     }
 957     downstream->pop_downstream_connection();
 958     // dconn was deleted
 959     dconn = nullptr;
 960   } else {
 961     auto rv = downstream->on_read();
 962     if (rv == SHRPX_ERR_EOF) {
 963       if (downstream->get_request_header_sent()) {
 964         return downstream_eof(dconn);
 965       }
 966       return SHRPX_ERR_RETRY;
 967     }
 968     if (rv == SHRPX_ERR_DCONN_CANCELED) {
 969       downstream->pop_downstream_connection();
 970       handler_->signal_write();
 971       return 0;
 972     }
 973     if (rv != 0) {
 974       if (rv != SHRPX_ERR_NETWORK) {
 975         if (LOG_ENABLED(INFO)) {
 976           DCLOG(INFO, dconn) << "HTTP parser failure";
 977         }
 978       }
 979       return downstream_error(dconn, Downstream::EVENT_ERROR);
 980     }
 981
 982     if (downstream->can_detach_downstream_connection()) {
 983       // Keep-alive
 984       downstream->detach_downstream_connection();
 985     }
 986   }
 987
 988   handler_->signal_write();
 989
 990   // At this point, downstream may be deleted.
 991
 992   return 0;
 993 }
 994
 995 int Http3Upstream::downstream_write(DownstreamConnection *dconn) {
 996   int rv;
 997   rv = dconn->on_write();
 998   if (rv == SHRPX_ERR_NETWORK) {
 999     return downstream_error(dconn, Downstream::EVENT_ERROR);
1000   }
1001   if (rv != 0) {
1002     return rv;
1003   }
1004   return 0;
1005 }
1006
1007 int Http3Upstream::downstream_eof(DownstreamConnection *dconn) {
1008   auto downstream = dconn->get_downstream();
1009
1010   if (LOG_ENABLED(INFO)) {
1011     DCLOG(INFO, dconn) << "EOF. stream_id=" << downstream->get_stream_id();
1012   }
1013
1014   // Delete downstream connection. If we don't delete it here, it will
1015   // be pooled in on_stream_close_callback.
1016   downstream->pop_downstream_connection();
1017   // dconn was deleted
1018   dconn = nullptr;
1019   // downstream wil be deleted in on_stream_close_callback.
1020   if (downstream->get_response_state() == DownstreamState::HEADER_COMPLETE) {
1021     // Server may indicate the end of the request by EOF
1022     if (LOG_ENABLED(INFO)) {
1023       ULOG(INFO, this) << "Downstream body was ended by EOF";
1024     }
1025     downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1026
1027     // For tunneled connection, MSG_COMPLETE signals
1028     // downstream_read_data_callback to send RST_STREAM after pending
1029     // response body is sent. This is needed to ensure that RST_STREAM
1030     // is sent after all pending data are sent.
1031     on_downstream_body_complete(downstream);
1032   } else if (downstream->get_response_state() !=
1033              DownstreamState::MSG_COMPLETE) {
1034     // If stream was not closed, then we set MSG_COMPLETE and let
1035     // on_stream_close_callback delete downstream.
1036     if (error_reply(downstream, 502) != 0) {
1037       return -1;
1038     }
1039   }
1040   handler_->signal_write();
1041   // At this point, downstream may be deleted.
1042   return 0;
1043 }
1044
1045 int Http3Upstream::downstream_error(DownstreamConnection *dconn, int events) {
1046   auto downstream = dconn->get_downstream();
1047
1048   if (LOG_ENABLED(INFO)) {
1049     if (events & Downstream::EVENT_ERROR) {
1050       DCLOG(INFO, dconn) << "Downstream network/general error";
1051     } else {
1052       DCLOG(INFO, dconn) << "Timeout";
1053     }
1054     if (downstream->get_upgraded()) {
1055       DCLOG(INFO, dconn) << "Note: this is tunnel connection";
1056     }
1057   }
1058
1059   // Delete downstream connection. If we don't delete it here, it will
1060   // be pooled in on_stream_close_callback.
1061   downstream->pop_downstream_connection();
1062   // dconn was deleted
1063   dconn = nullptr;
1064
1065   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1066     // For SSL tunneling, we issue RST_STREAM. For other types of
1067     // stream, we don't have to do anything since response was
1068     // complete.
1069     if (downstream->get_upgraded()) {
1070       shutdown_stream(downstream, NGHTTP3_H3_NO_ERROR);
1071     }
1072   } else {
1073     if (downstream->get_response_state() == DownstreamState::HEADER_COMPLETE) {
1074       if (downstream->get_upgraded()) {
1075         on_downstream_body_complete(downstream);
1076       } else {
1077         shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
1078       }
1079     } else {
1080       unsigned int status;
1081       if (events & Downstream::EVENT_TIMEOUT) {
1082         if (downstream->get_request_header_sent()) {
1083           status = 504;
1084         } else {
1085           status = 408;
1086         }
1087       } else {
1088         status = 502;
1089       }
1090       if (error_reply(downstream, status) != 0) {
1091         return -1;
1092       }
1093     }
1094     downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1095   }
1096   handler_->signal_write();
1097   // At this point, downstream may be deleted.
1098   return 0;
1099 }
1100
1101 ClientHandler *Http3Upstream::get_client_handler() const { return handler_; }
1102
1103 namespace {
1104 nghttp3_ssize downstream_read_data_callback(nghttp3_conn *conn,
1105                                             int64_t stream_id, nghttp3_vec *vec,
1106                                             size_t veccnt, uint32_t *pflags,
1107                                             void *conn_user_data,
1108                                             void *stream_user_data) {
1109   auto upstream = static_cast<Http3Upstream *>(conn_user_data);
1110   auto downstream = static_cast<Downstream *>(stream_user_data);
1111
1112   assert(downstream);
1113
1114   auto body = downstream->get_response_buf();
1115
1116   assert(body);
1117
1118   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1119     *pflags |= NGHTTP3_DATA_FLAG_EOF;
1120   } else if (body->rleft_mark() == 0) {
1121     downstream->disable_upstream_wtimer();
1122     return NGHTTP3_ERR_WOULDBLOCK;
1123   }
1124
1125   downstream->reset_upstream_wtimer();
1126
1127   veccnt = body->riovec_mark(reinterpret_cast<struct iovec *>(vec), veccnt);
1128
1129   assert((*pflags & NGHTTP3_DATA_FLAG_EOF) || veccnt);
1130
1131   downstream->response_sent_body_length += nghttp3_vec_len(vec, veccnt);
1132
1133   if ((*pflags & NGHTTP3_DATA_FLAG_EOF) &&
1134       upstream->shutdown_stream_read(stream_id, NGHTTP3_H3_NO_ERROR) != 0) {
1135     return NGHTTP3_ERR_CALLBACK_FAILURE;
1136   }
1137
1138   return veccnt;
1139 }
1140 } // namespace
1141
1142 int Http3Upstream::on_downstream_header_complete(Downstream *downstream) {
1143   int rv;
1144
1145   const auto &req = downstream->request();
1146   auto &resp = downstream->response();
1147
1148   auto &balloc = downstream->get_block_allocator();
1149
1150   if (LOG_ENABLED(INFO)) {
1151     if (downstream->get_non_final_response()) {
1152       DLOG(INFO, downstream) << "HTTP non-final response header";
1153     } else {
1154       DLOG(INFO, downstream) << "HTTP response header completed";
1155     }
1156   }
1157
1158   auto config = get_config();
1159   auto &httpconf = config->http;
1160
1161   if (!config->http2_proxy && !httpconf.no_location_rewrite) {
1162     downstream->rewrite_location_response_header(req.scheme);
1163   }
1164
1165 #ifdef HAVE_MRUBY
1166   if (!downstream->get_non_final_response()) {
1167     auto dconn = downstream->get_downstream_connection();
1168     const auto &group = dconn->get_downstream_addr_group();
1169     if (group) {
1170       const auto &dmruby_ctx = group->shared_addr->mruby_ctx;
1171
1172       if (dmruby_ctx->run_on_response_proc(downstream) != 0) {
1173         if (error_reply(downstream, 500) != 0) {
1174           return -1;
1175         }
1176         // Returning -1 will signal deletion of dconn.
1177         return -1;
1178       }
1179
1180       if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1181         return -1;
1182       }
1183     }
1184
1185     auto worker = handler_->get_worker();
1186     auto mruby_ctx = worker->get_mruby_context();
1187
1188     if (mruby_ctx->run_on_response_proc(downstream) != 0) {
1189       if (error_reply(downstream, 500) != 0) {
1190         return -1;
1191       }
1192       // Returning -1 will signal deletion of dconn.
1193       return -1;
1194     }
1195
1196     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1197       return -1;
1198     }
1199   }
1200 #endif // HAVE_MRUBY
1201
1202   auto nva = std::vector<nghttp3_nv>();
1203   // 4 means :status and possible server, via, and set-cookie (for
1204   // affinity cookie) header field.
1205   nva.reserve(resp.fs.headers().size() + 4 +
1206               httpconf.add_response_headers.size());
1207
1208   if (downstream->get_non_final_response()) {
1209     auto response_status = http2::stringify_status(balloc, resp.http_status);
1210
1211     nva.push_back(http3::make_nv_ls_nocopy(":status", response_status));
1212
1213     http3::copy_headers_to_nva_nocopy(nva, resp.fs.headers(),
1214                                       http2::HDOP_STRIP_ALL);
1215
1216     if (LOG_ENABLED(INFO)) {
1217       log_response_headers(downstream, nva);
1218     }
1219
1220     rv = nghttp3_conn_submit_info(httpconn_, downstream->get_stream_id(),
1221                                   nva.data(), nva.size());
1222
1223     resp.fs.clear_headers();
1224
1225     if (rv != 0) {
1226       ULOG(FATAL, this) << "nghttp3_conn_submit_info() failed";
1227       return -1;
1228     }
1229
1230     return 0;
1231   }
1232
1233   auto striphd_flags = http2::HDOP_STRIP_ALL & ~http2::HDOP_STRIP_VIA;
1234   StringRef response_status;
1235
1236   if (req.connect_proto == ConnectProto::WEBSOCKET && resp.http_status == 101) {
1237     response_status = http2::stringify_status(balloc, 200);
1238     striphd_flags |= http2::HDOP_STRIP_SEC_WEBSOCKET_ACCEPT;
1239   } else {
1240     response_status = http2::stringify_status(balloc, resp.http_status);
1241   }
1242
1243   nva.push_back(http3::make_nv_ls_nocopy(":status", response_status));
1244
1245   http3::copy_headers_to_nva_nocopy(nva, resp.fs.headers(), striphd_flags);
1246
1247   if (!config->http2_proxy && !httpconf.no_server_rewrite) {
1248     nva.push_back(http3::make_nv_ls_nocopy("server", httpconf.server_name));
1249   } else {
1250     auto server = resp.fs.header(http2::HD_SERVER);
1251     if (server) {
1252       nva.push_back(http3::make_nv_ls_nocopy("server", (*server).value));
1253     }
1254   }
1255
1256   if (!req.regular_connect_method() || !downstream->get_upgraded()) {
1257     auto affinity_cookie = downstream->get_affinity_cookie_to_send();
1258     if (affinity_cookie) {
1259       auto dconn = downstream->get_downstream_connection();
1260       assert(dconn);
1261       auto &group = dconn->get_downstream_addr_group();
1262       auto &shared_addr = group->shared_addr;
1263       auto &cookieconf = shared_addr->affinity.cookie;
1264       auto secure =
1265           http::require_cookie_secure_attribute(cookieconf.secure, req.scheme);
1266       auto cookie_str = http::create_affinity_cookie(
1267           balloc, cookieconf.name, affinity_cookie, cookieconf.path, secure);
1268       nva.push_back(http3::make_nv_ls_nocopy("set-cookie", cookie_str));
1269     }
1270   }
1271
1272   auto via = resp.fs.header(http2::HD_VIA);
1273   if (httpconf.no_via) {
1274     if (via) {
1275       nva.push_back(http3::make_nv_ls_nocopy("via", (*via).value));
1276     }
1277   } else {
1278     // we don't create more than 16 bytes in
1279     // http::create_via_header_value.
1280     size_t len = 16;
1281     if (via) {
1282       len += via->value.size() + 2;
1283     }
1284
1285     auto iov = make_byte_ref(balloc, len + 1);
1286     auto p = iov.base;
1287     if (via) {
1288       p = std::copy(std::begin(via->value), std::end(via->value), p);
1289       p = util::copy_lit(p, ", ");
1290     }
1291     p = http::create_via_header_value(p, resp.http_major, resp.http_minor);
1292     *p = '\0';
1293
1294     nva.push_back(http3::make_nv_ls_nocopy("via", StringRef{iov.base, p}));
1295   }
1296
1297   for (auto &p : httpconf.add_response_headers) {
1298     nva.push_back(http3::make_nv_nocopy(p.name, p.value));
1299   }
1300
1301   if (LOG_ENABLED(INFO)) {
1302     log_response_headers(downstream, nva);
1303   }
1304
1305   nghttp3_data_reader data_read;
1306   data_read.read_data = downstream_read_data_callback;
1307
1308   nghttp3_data_reader *data_readptr;
1309
1310   if (downstream->expect_response_body() ||
1311       downstream->expect_response_trailer()) {
1312     data_readptr = &data_read;
1313   } else {
1314     data_readptr = nullptr;
1315   }
1316
1317   rv = nghttp3_conn_submit_response(httpconn_, downstream->get_stream_id(),
1318                                     nva.data(), nva.size(), data_readptr);
1319   if (rv != 0) {
1320     ULOG(FATAL, this) << "nghttp3_conn_submit_response() failed";
1321     return -1;
1322   }
1323
1324   if (data_readptr) {
1325     downstream->reset_upstream_wtimer();
1326   } else if (shutdown_stream_read(downstream->get_stream_id(),
1327                                   NGHTTP3_H3_NO_ERROR) != 0) {
1328     return -1;
1329   }
1330
1331   return 0;
1332 }
1333
1334 int Http3Upstream::on_downstream_body(Downstream *downstream,
1335                                       const uint8_t *data, size_t len,
1336                                       bool flush) {
1337   auto body = downstream->get_response_buf();
1338   body->append(data, len);
1339
1340   if (flush) {
1341     nghttp3_conn_resume_stream(httpconn_, downstream->get_stream_id());
1342
1343     downstream->ensure_upstream_wtimer();
1344   }
1345
1346   return 0;
1347 }
1348
1349 int Http3Upstream::on_downstream_body_complete(Downstream *downstream) {
1350   if (LOG_ENABLED(INFO)) {
1351     DLOG(INFO, downstream) << "HTTP response completed";
1352   }
1353
1354   auto &resp = downstream->response();
1355
1356   if (!downstream->validate_response_recv_body_length()) {
1357     shutdown_stream(downstream, NGHTTP3_H3_GENERAL_PROTOCOL_ERROR);
1358     resp.connection_close = true;
1359     return 0;
1360   }
1361
1362   nghttp3_conn_resume_stream(httpconn_, downstream->get_stream_id());
1363   downstream->ensure_upstream_wtimer();
1364
1365   return 0;
1366 }
1367
1368 void Http3Upstream::on_handler_delete() {
1369   for (auto d = downstream_queue_.get_downstreams(); d; d = d->dlnext) {
1370     if (d->get_dispatch_state() == DispatchState::ACTIVE &&
1371         d->accesslog_ready()) {
1372       handler_->write_accesslog(d);
1373     }
1374   }
1375
1376   auto worker = handler_->get_worker();
1377   auto quic_conn_handler = worker->get_quic_connection_handler();
1378
1379   std::vector<ngtcp2_cid> scids(ngtcp2_conn_get_num_scid(conn_) + 1);
1380   ngtcp2_conn_get_scid(conn_, scids.data());
1381   scids.back() = hashed_scid_;
1382
1383   for (auto &cid : scids) {
1384     quic_conn_handler->remove_connection_id(cid);
1385   }
1386
1387   if (idle_close_ || retry_close_) {
1388     return;
1389   }
1390
1391   // If this is not idle close, send CONNECTION_CLOSE.
1392   if (!ngtcp2_conn_is_in_closing_period(conn_) &&
1393       !ngtcp2_conn_is_in_draining_period(conn_)) {
1394     ngtcp2_path_storage ps;
1395     ngtcp2_pkt_info pi;
1396     conn_close_.resize(SHRPX_QUIC_CONN_CLOSE_PKTLEN);
1397
1398     ngtcp2_path_storage_zero(&ps);
1399
1400     auto nwrite = ngtcp2_conn_write_connection_close(
1401         conn_, &ps.path, &pi, conn_close_.data(), conn_close_.size(),
1402         NGTCP2_NO_ERROR, quic_timestamp());
1403     if (nwrite < 0) {
1404       if (nwrite != NGTCP2_ERR_INVALID_STATE) {
1405         ULOG(ERROR, this) << "ngtcp2_conn_write_connection_close: "
1406                           << ngtcp2_strerror(nwrite);
1407       }
1408
1409       return;
1410     }
1411
1412     conn_close_.resize(nwrite);
1413
1414     send_packet(static_cast<UpstreamAddr *>(ps.path.user_data),
1415                 ps.path.remote.addr, ps.path.remote.addrlen, ps.path.local.addr,
1416                 ps.path.local.addrlen, conn_close_.data(), nwrite, 0);
1417   }
1418
1419   auto d =
1420       static_cast<ev_tstamp>(ngtcp2_conn_get_pto(conn_) * 3) / NGTCP2_SECONDS;
1421
1422   if (LOG_ENABLED(INFO)) {
1423     ULOG(INFO, this) << "Enter close-wait period " << d << "s with "
1424                      << conn_close_.size() << " bytes sentinel packet";
1425   }
1426
1427   auto cw = std::make_unique<CloseWait>(worker, std::move(scids),
1428                                         std::move(conn_close_), d);
1429
1430   quic_conn_handler->add_close_wait(cw.get());
1431
1432   cw.release();
1433 }
1434
1435 int Http3Upstream::on_downstream_reset(Downstream *downstream, bool no_retry) {
1436   int rv;
1437
1438   if (downstream->get_dispatch_state() != DispatchState::ACTIVE) {
1439     // This is error condition when we failed push_request_headers()
1440     // in initiate_downstream().  Otherwise, we have
1441     // DispatchState::ACTIVE state, or we did not set
1442     // DownstreamConnection.
1443     downstream->pop_downstream_connection();
1444     handler_->signal_write();
1445
1446     return 0;
1447   }
1448
1449   if (!downstream->request_submission_ready()) {
1450     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1451       // We have got all response body already.  Send it off.
1452       downstream->pop_downstream_connection();
1453       return 0;
1454     }
1455     // pushed stream is handled here
1456     shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
1457     downstream->pop_downstream_connection();
1458
1459     handler_->signal_write();
1460
1461     return 0;
1462   }
1463
1464   downstream->pop_downstream_connection();
1465
1466   downstream->add_retry();
1467
1468   std::unique_ptr<DownstreamConnection> dconn;
1469
1470   rv = 0;
1471
1472   if (no_retry || downstream->no_more_retry()) {
1473     goto fail;
1474   }
1475
1476   // downstream connection is clean; we can retry with new
1477   // downstream connection.
1478
1479   for (;;) {
1480     auto dconn = handler_->get_downstream_connection(rv, downstream);
1481     if (!dconn) {
1482       goto fail;
1483     }
1484
1485     rv = downstream->attach_downstream_connection(std::move(dconn));
1486     if (rv == 0) {
1487       break;
1488     }
1489   }
1490
1491   rv = downstream->push_request_headers();
1492   if (rv != 0) {
1493     goto fail;
1494   }
1495
1496   return 0;
1497
1498 fail:
1499   if (rv == SHRPX_ERR_TLS_REQUIRED) {
1500     rv = on_downstream_abort_request_with_https_redirect(downstream);
1501   } else {
1502     rv = on_downstream_abort_request(downstream, 502);
1503   }
1504   if (rv != 0) {
1505     shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
1506   }
1507   downstream->pop_downstream_connection();
1508
1509   handler_->signal_write();
1510
1511   return 0;
1512 }
1513
1514 void Http3Upstream::pause_read(IOCtrlReason reason) {}
1515
1516 int Http3Upstream::resume_read(IOCtrlReason reason, Downstream *downstream,
1517                                size_t consumed) {
1518   consume(downstream->get_stream_id(), consumed);
1519
1520   auto &req = downstream->request();
1521
1522   req.consume(consumed);
1523
1524   handler_->signal_write();
1525
1526   return 0;
1527 }
1528
1529 int Http3Upstream::send_reply(Downstream *downstream, const uint8_t *body,
1530                               size_t bodylen) {
1531   int rv;
1532
1533   nghttp3_data_reader data_read, *data_read_ptr = nullptr;
1534
1535   if (bodylen) {
1536     data_read.read_data = downstream_read_data_callback;
1537     data_read_ptr = &data_read;
1538   }
1539
1540   const auto &resp = downstream->response();
1541   auto config = get_config();
1542   auto &httpconf = config->http;
1543
1544   auto &balloc = downstream->get_block_allocator();
1545
1546   const auto &headers = resp.fs.headers();
1547   auto nva = std::vector<nghttp3_nv>();
1548   // 2 for :status and server
1549   nva.reserve(2 + headers.size() + httpconf.add_response_headers.size());
1550
1551   auto response_status = http2::stringify_status(balloc, resp.http_status);
1552
1553   nva.push_back(http3::make_nv_ls_nocopy(":status", response_status));
1554
1555   for (auto &kv : headers) {
1556     if (kv.name.empty() || kv.name[0] == ':') {
1557       continue;
1558     }
1559     switch (kv.token) {
1560     case http2::HD_CONNECTION:
1561     case http2::HD_KEEP_ALIVE:
1562     case http2::HD_PROXY_CONNECTION:
1563     case http2::HD_TE:
1564     case http2::HD_TRANSFER_ENCODING:
1565     case http2::HD_UPGRADE:
1566       continue;
1567     }
1568     nva.push_back(http3::make_nv_nocopy(kv.name, kv.value, kv.no_index));
1569   }
1570
1571   if (!resp.fs.header(http2::HD_SERVER)) {
1572     nva.push_back(http3::make_nv_ls_nocopy("server", config->http.server_name));
1573   }
1574
1575   for (auto &p : httpconf.add_response_headers) {
1576     nva.push_back(http3::make_nv_nocopy(p.name, p.value));
1577   }
1578
1579   rv = nghttp3_conn_submit_response(httpconn_, downstream->get_stream_id(),
1580                                     nva.data(), nva.size(), data_read_ptr);
1581   if (nghttp3_err_is_fatal(rv)) {
1582     ULOG(FATAL, this) << "nghttp3_conn_submit_response() failed: "
1583                       << nghttp3_strerror(rv);
1584     return -1;
1585   }
1586
1587   auto buf = downstream->get_response_buf();
1588
1589   buf->append(body, bodylen);
1590
1591   downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1592
1593   if (data_read_ptr) {
1594     downstream->reset_upstream_wtimer();
1595   }
1596
1597   if (shutdown_stream_read(downstream->get_stream_id(), NGHTTP3_H3_NO_ERROR) !=
1598       0) {
1599     return -1;
1600   }
1601
1602   return 0;
1603 }
1604
1605 int Http3Upstream::initiate_push(Downstream *downstream, const StringRef &uri) {
1606   return 0;
1607 }
1608
1609 int Http3Upstream::response_riovec(struct iovec *iov, int iovcnt) const {
1610   return 0;
1611 }
1612
1613 void Http3Upstream::response_drain(size_t n) {}
1614
1615 bool Http3Upstream::response_empty() const { return false; }
1616
1617 Downstream *
1618 Http3Upstream::on_downstream_push_promise(Downstream *downstream,
1619                                           int32_t promised_stream_id) {
1620   return nullptr;
1621 }
1622
1623 int Http3Upstream::on_downstream_push_promise_complete(
1624     Downstream *downstream, Downstream *promised_downstream) {
1625   return 0;
1626 }
1627
1628 bool Http3Upstream::push_enabled() const { return false; }
1629
1630 void Http3Upstream::cancel_premature_downstream(
1631     Downstream *promised_downstream) {}
1632
1633 int Http3Upstream::on_read(const UpstreamAddr *faddr,
1634                            const Address &remote_addr,
1635                            const Address &local_addr, const uint8_t *data,
1636                            size_t datalen) {
1637   int rv;
1638   ngtcp2_pkt_info pi{};
1639
1640   auto path = ngtcp2_path{
1641       {
1642           local_addr.len,
1643           const_cast<sockaddr *>(&local_addr.su.sa),
1644       },
1645       {
1646           remote_addr.len,
1647           const_cast<sockaddr *>(&remote_addr.su.sa),
1648       },
1649       const_cast<UpstreamAddr *>(faddr),
1650   };
1651
1652   rv = ngtcp2_conn_read_pkt(conn_, &path, &pi, data, datalen, quic_timestamp());
1653   if (rv != 0) {
1654     switch (rv) {
1655     case NGTCP2_ERR_DRAINING:
1656       return -1;
1657     case NGTCP2_ERR_RETRY: {
1658       auto worker = handler_->get_worker();
1659       auto quic_conn_handler = worker->get_quic_connection_handler();
1660
1661       uint32_t version;
1662       const uint8_t *dcid, *scid;
1663       size_t dcidlen, scidlen;
1664
1665       rv = ngtcp2_pkt_decode_version_cid(&version, &dcid, &dcidlen, &scid,
1666                                          &scidlen, data, datalen,
1667                                          SHRPX_QUIC_SCIDLEN);
1668       if (rv != 0) {
1669         return -1;
1670       }
1671
1672       if (worker->get_graceful_shutdown()) {
1673         ngtcp2_cid ini_dcid, ini_scid;
1674
1675         ngtcp2_cid_init(&ini_dcid, dcid, dcidlen);
1676         ngtcp2_cid_init(&ini_scid, scid, scidlen);
1677
1678         quic_conn_handler->send_connection_close(
1679             faddr, version, ini_dcid, ini_scid, remote_addr, local_addr,
1680             NGTCP2_CONNECTION_REFUSED);
1681
1682         return -1;
1683       }
1684
1685       retry_close_ = true;
1686
1687       quic_conn_handler->send_retry(handler_->get_upstream_addr(), version,
1688                                     dcid, dcidlen, scid, scidlen, remote_addr,
1689                                     local_addr);
1690
1691       return -1;
1692     }
1693     case NGTCP2_ERR_REQUIRED_TRANSPORT_PARAM:
1694     case NGTCP2_ERR_MALFORMED_TRANSPORT_PARAM:
1695     case NGTCP2_ERR_TRANSPORT_PARAM:
1696       // If rv indicates transport_parameters related error, we should
1697       // send TRANSPORT_PARAMETER_ERROR even if last_error_.code is
1698       // already set.  This is because OpenSSL might set Alert.
1699       last_error_ = quic::err_transport(rv);
1700       break;
1701     case NGTCP2_ERR_DROP_CONN:
1702       return -1;
1703     default:
1704       if (!last_error_.code) {
1705         last_error_ = quic::err_transport(rv);
1706       }
1707     }
1708
1709     ULOG(ERROR, this) << "ngtcp2_conn_read_pkt: " << ngtcp2_strerror(rv);
1710
1711     return handle_error();
1712   }
1713
1714   reset_idle_timer();
1715
1716   return 0;
1717 }
1718
1719 int Http3Upstream::send_packet(const UpstreamAddr *faddr,
1720                                const sockaddr *remote_sa, size_t remote_salen,
1721                                const sockaddr *local_sa, size_t local_salen,
1722                                const uint8_t *data, size_t datalen,
1723                                size_t gso_size) {
1724   auto rv = quic_send_packet(faddr, remote_sa, remote_salen, local_sa,
1725                              local_salen, data, datalen, gso_size);
1726   switch (rv) {
1727   case 0:
1728     return 0;
1729     // With GSO, sendmsg may fail with EINVAL if UDP payload is too
1730     // large.
1731   case -EINVAL:
1732   case -EMSGSIZE:
1733     max_udp_payload_size_ = NGTCP2_MAX_UDP_PAYLOAD_SIZE;
1734     break;
1735   default:
1736     break;
1737   }
1738
1739   return -1;
1740 }
1741
1742 int Http3Upstream::handle_error() {
1743   if (ngtcp2_conn_is_in_closing_period(conn_)) {
1744     return -1;
1745   }
1746
1747   ngtcp2_path_storage ps;
1748   ngtcp2_pkt_info pi;
1749
1750   ngtcp2_path_storage_zero(&ps);
1751
1752   auto ts = quic_timestamp();
1753
1754   conn_close_.resize(SHRPX_QUIC_CONN_CLOSE_PKTLEN);
1755
1756   ngtcp2_ssize nwrite;
1757
1758   if (last_error_.type == quic::ErrorType::Transport) {
1759     nwrite = ngtcp2_conn_write_connection_close(
1760         conn_, &ps.path, &pi, conn_close_.data(), conn_close_.size(),
1761         last_error_.code, ts);
1762     if (nwrite < 0) {
1763       ULOG(ERROR, this) << "ngtcp2_conn_write_connection_close: "
1764                         << ngtcp2_strerror(nwrite);
1765       return -1;
1766     }
1767   } else {
1768     nwrite = ngtcp2_conn_write_application_close(
1769         conn_, &ps.path, &pi, conn_close_.data(), conn_close_.size(),
1770         last_error_.code, ts);
1771     if (nwrite < 0) {
1772       ULOG(ERROR, this) << "ngtcp2_conn_write_application_close: "
1773                         << ngtcp2_strerror(nwrite);
1774       return -1;
1775     }
1776   }
1777
1778   conn_close_.resize(nwrite);
1779
1780   send_packet(static_cast<UpstreamAddr *>(ps.path.user_data),
1781               ps.path.remote.addr, ps.path.remote.addrlen, ps.path.local.addr,
1782               ps.path.local.addrlen, conn_close_.data(), nwrite, 0);
1783
1784   return -1;
1785 }
1786
1787 int Http3Upstream::on_rx_secret(ngtcp2_crypto_level level,
1788                                 const uint8_t *secret, size_t secretlen) {
1789   if (ngtcp2_crypto_derive_and_install_rx_key(conn_, nullptr, nullptr, nullptr,
1790                                               level, secret, secretlen) != 0) {
1791     ULOG(ERROR, this) << "ngtcp2_crypto_derive_and_install_rx_key failed";
1792     return -1;
1793   }
1794
1795   return 0;
1796 }
1797
1798 int Http3Upstream::on_tx_secret(ngtcp2_crypto_level level,
1799                                 const uint8_t *secret, size_t secretlen) {
1800   if (ngtcp2_crypto_derive_and_install_tx_key(conn_, nullptr, nullptr, nullptr,
1801                                               level, secret, secretlen) != 0) {
1802     ULOG(ERROR, this) << "ngtcp2_crypto_derive_and_install_tx_key failed";
1803     return -1;
1804   }
1805
1806   if (level == NGTCP2_CRYPTO_LEVEL_APPLICATION && setup_httpconn() != 0) {
1807     return -1;
1808   }
1809
1810   return 0;
1811 }
1812
1813 int Http3Upstream::add_crypto_data(ngtcp2_crypto_level level,
1814                                    const uint8_t *data, size_t datalen) {
1815   int rv = ngtcp2_conn_submit_crypto_data(conn_, level, data, datalen);
1816
1817   if (rv != 0) {
1818     ULOG(ERROR, this) << "ngtcp2_conn_submit_crypto_data: "
1819                       << ngtcp2_strerror(rv);
1820     return -1;
1821   }
1822
1823   return 0;
1824 }
1825
1826 void Http3Upstream::set_tls_alert(uint8_t alert) { tls_alert_ = alert; }
1827
1828 int Http3Upstream::handle_expiry() {
1829   int rv;
1830
1831   auto ts = quic_timestamp();
1832
1833   rv = ngtcp2_conn_handle_expiry(conn_, ts);
1834   if (rv != 0) {
1835     ULOG(ERROR, this) << "ngtcp2_conn_handle_expiry: " << ngtcp2_strerror(rv);
1836     last_error_ = quic::err_transport(rv);
1837     return handle_error();
1838   }
1839
1840   return 0;
1841 }
1842
1843 void Http3Upstream::reset_idle_timer() {
1844   auto ts = quic_timestamp();
1845   auto idle_ts = ngtcp2_conn_get_idle_expiry(conn_);
1846
1847   idle_timer_.repeat =
1848       idle_ts > ts ? static_cast<ev_tstamp>(idle_ts - ts) / NGTCP2_SECONDS
1849                    : 1e-9;
1850
1851   ev_timer_again(handler_->get_loop(), &idle_timer_);
1852 }
1853
1854 void Http3Upstream::reset_timer() {
1855   auto ts = quic_timestamp();
1856   auto expiry_ts = ngtcp2_conn_get_expiry(conn_);
1857
1858   timer_.repeat = expiry_ts > ts
1859                       ? static_cast<ev_tstamp>(expiry_ts - ts) / NGTCP2_SECONDS
1860                       : 1e-9;
1861
1862   ev_timer_again(handler_->get_loop(), &timer_);
1863 }
1864
1865 namespace {
1866 int http_deferred_consume(nghttp3_conn *conn, int64_t stream_id,
1867                           size_t nconsumed, void *user_data,
1868                           void *stream_user_data) {
1869   auto upstream = static_cast<Http3Upstream *>(user_data);
1870
1871   upstream->consume(stream_id, nconsumed);
1872
1873   return 0;
1874 }
1875 } // namespace
1876
1877 namespace {
1878 int http_acked_stream_data(nghttp3_conn *conn, int64_t stream_id,
1879                            uint64_t datalen, void *user_data,
1880                            void *stream_user_data) {
1881   auto upstream = static_cast<Http3Upstream *>(user_data);
1882   auto downstream = static_cast<Downstream *>(stream_user_data);
1883
1884   assert(downstream);
1885
1886   if (upstream->http_acked_stream_data(downstream, datalen) != 0) {
1887     return NGHTTP3_ERR_CALLBACK_FAILURE;
1888   }
1889
1890   return 0;
1891 }
1892 } // namespace
1893
1894 int Http3Upstream::http_acked_stream_data(Downstream *downstream,
1895                                           uint64_t datalen) {
1896   if (LOG_ENABLED(INFO)) {
1897     ULOG(INFO, this) << "Stream " << downstream->get_stream_id() << " "
1898                      << datalen << " bytes acknowledged";
1899   }
1900
1901   auto body = downstream->get_response_buf();
1902   auto drained = body->drain_mark(datalen);
1903   (void)drained;
1904
1905   assert(datalen == drained);
1906
1907   if (downstream->resume_read(SHRPX_NO_BUFFER, datalen) != 0) {
1908     return -1;
1909   }
1910
1911   return 0;
1912 }
1913
1914 namespace {
1915 int http_begin_request_headers(nghttp3_conn *conn, int64_t stream_id,
1916                                void *user_data, void *stream_user_data) {
1917   if (!ngtcp2_is_bidi_stream(stream_id)) {
1918     return 0;
1919   }
1920
1921   auto upstream = static_cast<Http3Upstream *>(user_data);
1922   upstream->http_begin_request_headers(stream_id);
1923
1924   return 0;
1925 }
1926 } // namespace
1927
1928 namespace {
1929 int http_recv_request_header(nghttp3_conn *conn, int64_t stream_id,
1930                              int32_t token, nghttp3_rcbuf *name,
1931                              nghttp3_rcbuf *value, uint8_t flags,
1932                              void *user_data, void *stream_user_data) {
1933   auto upstream = static_cast<Http3Upstream *>(user_data);
1934   auto downstream = static_cast<Downstream *>(stream_user_data);
1935
1936   if (!downstream || downstream->get_stop_reading()) {
1937     return 0;
1938   }
1939
1940   if (upstream->http_recv_request_header(downstream, token, name, value,
1941                                          flags) != 0) {
1942     return NGHTTP3_ERR_CALLBACK_FAILURE;
1943   }
1944
1945   return 0;
1946 }
1947 } // namespace
1948
1949 int Http3Upstream::http_recv_request_header(Downstream *downstream,
1950                                             int32_t h3token,
1951                                             nghttp3_rcbuf *name,
1952                                             nghttp3_rcbuf *value,
1953                                             uint8_t flags) {
1954   auto namebuf = nghttp3_rcbuf_get_buf(name);
1955   auto valuebuf = nghttp3_rcbuf_get_buf(value);
1956   auto &req = downstream->request();
1957   auto config = get_config();
1958   auto &httpconf = config->http;
1959
1960   if (req.fs.buffer_size() + namebuf.len + valuebuf.len >
1961           httpconf.request_header_field_buffer ||
1962       req.fs.num_fields() >= httpconf.max_request_header_fields) {
1963     downstream->set_stop_reading(true);
1964
1965     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
1966       return 0;
1967     }
1968
1969     if (LOG_ENABLED(INFO)) {
1970       ULOG(INFO, this) << "Too large or many header field size="
1971                        << req.fs.buffer_size() + namebuf.len + valuebuf.len
1972                        << ", num=" << req.fs.num_fields() + 1;
1973     }
1974
1975     if (error_reply(downstream, 431) != 0) {
1976       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
1977     }
1978
1979     return 0;
1980   }
1981
1982   auto token = http2::lookup_token(namebuf.base, namebuf.len);
1983   auto no_index = flags & NGHTTP3_NV_FLAG_NEVER_INDEX;
1984
1985   downstream->add_rcbuf(name);
1986   downstream->add_rcbuf(value);
1987
1988   req.fs.add_header_token(StringRef{namebuf.base, namebuf.len},
1989                           StringRef{valuebuf.base, valuebuf.len}, no_index,
1990                           token);
1991   return 0;
1992 }
1993
1994 namespace {
1995 int http_end_request_headers(nghttp3_conn *conn, int64_t stream_id,
1996                              void *user_data, void *stream_user_data) {
1997   auto upstream = static_cast<Http3Upstream *>(user_data);
1998   auto handler = upstream->get_client_handler();
1999   auto downstream = static_cast<Downstream *>(stream_user_data);
2000
2001   if (!downstream || downstream->get_stop_reading()) {
2002     return 0;
2003   }
2004
2005   if (upstream->http_end_request_headers(downstream) != 0) {
2006     return NGHTTP3_ERR_CALLBACK_FAILURE;
2007   }
2008
2009   downstream->reset_upstream_rtimer();
2010   handler->stop_read_timer();
2011
2012   return 0;
2013 }
2014 } // namespace
2015
2016 int Http3Upstream::http_end_request_headers(Downstream *downstream) {
2017   auto lgconf = log_config();
2018   lgconf->update_tstamp(std::chrono::system_clock::now());
2019   auto &req = downstream->request();
2020   req.tstamp = lgconf->tstamp;
2021
2022   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
2023     return 0;
2024   }
2025
2026   auto &nva = req.fs.headers();
2027
2028   if (LOG_ENABLED(INFO)) {
2029     std::stringstream ss;
2030     for (auto &nv : nva) {
2031       if (nv.name == "authorization") {
2032         ss << TTY_HTTP_HD << nv.name << TTY_RST << ": <redacted>\n";
2033         continue;
2034       }
2035       ss << TTY_HTTP_HD << nv.name << TTY_RST << ": " << nv.value << "\n";
2036     }
2037     ULOG(INFO, this) << "HTTP request headers. stream_id="
2038                      << downstream->get_stream_id() << "\n"
2039                      << ss.str();
2040   }
2041
2042   auto content_length = req.fs.header(http2::HD_CONTENT_LENGTH);
2043   if (content_length) {
2044     // libnghttp2 guarantees this can be parsed
2045     req.fs.content_length = util::parse_uint(content_length->value);
2046   }
2047
2048   // presence of mandatory header fields are guaranteed by libnghttp2.
2049   auto authority = req.fs.header(http2::HD__AUTHORITY);
2050   auto path = req.fs.header(http2::HD__PATH);
2051   auto method = req.fs.header(http2::HD__METHOD);
2052   auto scheme = req.fs.header(http2::HD__SCHEME);
2053
2054   auto method_token = http2::lookup_method_token(method->value);
2055   if (method_token == -1) {
2056     if (error_reply(downstream, 501) != 0) {
2057       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
2058     }
2059     return 0;
2060   }
2061
2062   auto faddr = handler_->get_upstream_addr();
2063
2064   auto config = get_config();
2065
2066   // For HTTP/2 proxy, we require :authority.
2067   if (method_token != HTTP_CONNECT && config->http2_proxy &&
2068       faddr->alt_mode == UpstreamAltMode::NONE && !authority) {
2069     shutdown_stream(downstream, NGHTTP2_PROTOCOL_ERROR);
2070     return 0;
2071   }
2072
2073   req.method = method_token;
2074   if (scheme) {
2075     req.scheme = scheme->value;
2076   }
2077
2078   // nghttp2 library guarantees either :authority or host exist
2079   if (!authority) {
2080     req.no_authority = true;
2081     authority = req.fs.header(http2::HD_HOST);
2082   }
2083
2084   if (authority) {
2085     req.authority = authority->value;
2086   }
2087
2088   if (path) {
2089     if (method_token == HTTP_OPTIONS &&
2090         path->value == StringRef::from_lit("*")) {
2091       // Server-wide OPTIONS request.  Path is empty.
2092     } else if (config->http2_proxy &&
2093                faddr->alt_mode == UpstreamAltMode::NONE) {
2094       req.path = path->value;
2095     } else {
2096       req.path = http2::rewrite_clean_path(downstream->get_block_allocator(),
2097                                            path->value);
2098     }
2099   }
2100
2101   auto connect_proto = req.fs.header(http2::HD__PROTOCOL);
2102   if (connect_proto) {
2103     if (connect_proto->value != "websocket") {
2104       if (error_reply(downstream, 400) != 0) {
2105         return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
2106       }
2107       return 0;
2108     }
2109     req.connect_proto = ConnectProto::WEBSOCKET;
2110   }
2111
2112   // We are not sure that request has body or not at the moment.
2113   req.http2_expect_body = true;
2114
2115   downstream->inspect_http2_request();
2116
2117   downstream->set_request_state(DownstreamState::HEADER_COMPLETE);
2118
2119 #ifdef HAVE_MRUBY
2120   auto upstream = downstream->get_upstream();
2121   auto handler = upstream->get_client_handler();
2122   auto worker = handler->get_worker();
2123   auto mruby_ctx = worker->get_mruby_context();
2124
2125   if (mruby_ctx->run_on_request_proc(downstream) != 0) {
2126     if (error_reply(downstream, 500) != 0) {
2127       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
2128     }
2129     return 0;
2130   }
2131 #endif // HAVE_MRUBY
2132
2133   if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
2134     return 0;
2135   }
2136
2137   start_downstream(downstream);
2138
2139   return 0;
2140 }
2141
2142 void Http3Upstream::start_downstream(Downstream *downstream) {
2143   if (downstream_queue_.can_activate(downstream->request().authority)) {
2144     initiate_downstream(downstream);
2145     return;
2146   }
2147
2148   downstream_queue_.mark_blocked(downstream);
2149 }
2150
2151 void Http3Upstream::initiate_downstream(Downstream *downstream) {
2152   int rv;
2153
2154   DownstreamConnection *dconn_ptr;
2155
2156   for (;;) {
2157     auto dconn = handler_->get_downstream_connection(rv, downstream);
2158     if (!dconn) {
2159       if (rv == SHRPX_ERR_TLS_REQUIRED) {
2160         rv = redirect_to_https(downstream);
2161       } else {
2162         rv = error_reply(downstream, 502);
2163       }
2164       if (rv != 0) {
2165         shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
2166       }
2167
2168       downstream->set_request_state(DownstreamState::CONNECT_FAIL);
2169       downstream_queue_.mark_failure(downstream);
2170
2171       return;
2172     }
2173
2174 #ifdef HAVE_MRUBY
2175     dconn_ptr = dconn.get();
2176 #endif // HAVE_MRUBY
2177     rv = downstream->attach_downstream_connection(std::move(dconn));
2178     if (rv == 0) {
2179       break;
2180     }
2181   }
2182
2183 #ifdef HAVE_MRUBY
2184   const auto &group = dconn_ptr->get_downstream_addr_group();
2185   if (group) {
2186     const auto &mruby_ctx = group->shared_addr->mruby_ctx;
2187     if (mruby_ctx->run_on_request_proc(downstream) != 0) {
2188       if (error_reply(downstream, 500) != 0) {
2189         shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
2190       }
2191
2192       downstream_queue_.mark_failure(downstream);
2193
2194       return;
2195     }
2196
2197     if (downstream->get_response_state() == DownstreamState::MSG_COMPLETE) {
2198       return;
2199     }
2200   }
2201 #endif // HAVE_MRUBY
2202
2203   rv = downstream->push_request_headers();
2204   if (rv != 0) {
2205
2206     if (error_reply(downstream, 502) != 0) {
2207       shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
2208     }
2209
2210     downstream_queue_.mark_failure(downstream);
2211
2212     return;
2213   }
2214
2215   downstream_queue_.mark_active(downstream);
2216
2217   auto &req = downstream->request();
2218   if (!req.http2_expect_body) {
2219     rv = downstream->end_upload_data();
2220     if (rv != 0) {
2221       shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
2222     }
2223   }
2224 }
2225
2226 namespace {
2227 int http_recv_data(nghttp3_conn *conn, int64_t stream_id, const uint8_t *data,
2228                    size_t datalen, void *user_data, void *stream_user_data) {
2229   auto upstream = static_cast<Http3Upstream *>(user_data);
2230   auto downstream = static_cast<Downstream *>(stream_user_data);
2231
2232   if (upstream->http_recv_data(downstream, data, datalen) != 0) {
2233     return NGHTTP3_ERR_CALLBACK_FAILURE;
2234   }
2235
2236   return 0;
2237 }
2238 } // namespace
2239
2240 int Http3Upstream::http_recv_data(Downstream *downstream, const uint8_t *data,
2241                                   size_t datalen) {
2242   downstream->reset_upstream_rtimer();
2243
2244   if (downstream->push_upload_data_chunk(data, datalen) != 0) {
2245     if (downstream->get_response_state() != DownstreamState::MSG_COMPLETE) {
2246       shutdown_stream(downstream, NGHTTP3_H3_INTERNAL_ERROR);
2247     }
2248
2249     consume(downstream->get_stream_id(), datalen);
2250
2251     return 0;
2252   }
2253
2254   return 0;
2255 }
2256
2257 namespace {
2258 int http_end_stream(nghttp3_conn *conn, int64_t stream_id, void *user_data,
2259                     void *stream_user_data) {
2260   auto upstream = static_cast<Http3Upstream *>(user_data);
2261   auto downstream = static_cast<Downstream *>(stream_user_data);
2262
2263   if (!downstream || downstream->get_stop_reading()) {
2264     return 0;
2265   }
2266
2267   if (upstream->http_end_stream(downstream) != 0) {
2268     return NGHTTP3_ERR_CALLBACK_FAILURE;
2269   }
2270
2271   return 0;
2272 }
2273 } // namespace
2274
2275 int Http3Upstream::http_end_stream(Downstream *downstream) {
2276   downstream->disable_upstream_rtimer();
2277
2278   if (downstream->end_upload_data() != 0) {
2279     if (downstream->get_response_state() != DownstreamState::MSG_COMPLETE) {
2280       shutdown_stream(downstream, NGHTTP2_INTERNAL_ERROR);
2281     }
2282   }
2283
2284   downstream->set_request_state(DownstreamState::MSG_COMPLETE);
2285
2286   return 0;
2287 }
2288
2289 namespace {
2290 int http_stream_close(nghttp3_conn *conn, int64_t stream_id,
2291                       uint64_t app_error_code, void *conn_user_data,
2292                       void *stream_user_data) {
2293   auto upstream = static_cast<Http3Upstream *>(conn_user_data);
2294   auto downstream = static_cast<Downstream *>(stream_user_data);
2295
2296   if (!downstream) {
2297     return 0;
2298   }
2299
2300   if (upstream->http_stream_close(downstream, app_error_code) != 0) {
2301     return NGHTTP3_ERR_CALLBACK_FAILURE;
2302   }
2303
2304   return 0;
2305 }
2306 } // namespace
2307
2308 int Http3Upstream::http_stream_close(Downstream *downstream,
2309                                      uint64_t app_error_code) {
2310   auto stream_id = downstream->get_stream_id();
2311
2312   if (LOG_ENABLED(INFO)) {
2313     ULOG(INFO, this) << "Stream stream_id=" << stream_id
2314                      << " is being closed with app_error_code="
2315                      << app_error_code;
2316
2317     auto body = downstream->get_response_buf();
2318
2319     ULOG(INFO, this) << "response unacked_left=" << body->rleft()
2320                      << " not_sent=" << body->rleft_mark();
2321   }
2322
2323   auto &req = downstream->request();
2324
2325   consume(stream_id, req.unconsumed_body_length);
2326
2327   req.unconsumed_body_length = 0;
2328
2329   ngtcp2_conn_extend_max_streams_bidi(conn_, 1);
2330
2331   if (downstream->get_request_state() == DownstreamState::CONNECT_FAIL) {
2332     remove_downstream(downstream);
2333     // downstream was deleted
2334
2335     return 0;
2336   }
2337
2338   if (downstream->can_detach_downstream_connection()) {
2339     // Keep-alive
2340     downstream->detach_downstream_connection();
2341   }
2342
2343   downstream->set_request_state(DownstreamState::STREAM_CLOSED);
2344
2345   // At this point, downstream read may be paused.
2346
2347   // If shrpx_downstream::push_request_headers() failed, the
2348   // error is handled here.
2349   remove_downstream(downstream);
2350   // downstream was deleted
2351
2352   return 0;
2353 }
2354
2355 namespace {
2356 int http_send_stop_sending(nghttp3_conn *conn, int64_t stream_id,
2357                            uint64_t app_error_code, void *user_data,
2358                            void *stream_user_data) {
2359   auto upstream = static_cast<Http3Upstream *>(user_data);
2360
2361   if (upstream->http_send_stop_sending(stream_id, app_error_code) != 0) {
2362     return NGHTTP3_ERR_CALLBACK_FAILURE;
2363   }
2364
2365   return 0;
2366 }
2367 } // namespace
2368
2369 int Http3Upstream::http_send_stop_sending(int64_t stream_id,
2370                                           uint64_t app_error_code) {
2371   auto rv = ngtcp2_conn_shutdown_stream_read(conn_, stream_id, app_error_code);
2372   if (ngtcp2_err_is_fatal(rv)) {
2373     ULOG(ERROR, this) << "ngtcp2_conn_shutdown_stream_read: "
2374                       << ngtcp2_strerror(rv);
2375     return -1;
2376   }
2377
2378   return 0;
2379 }
2380
2381 namespace {
2382 int http_reset_stream(nghttp3_conn *conn, int64_t stream_id,
2383                       uint64_t app_error_code, void *user_data,
2384                       void *stream_user_data) {
2385   auto upstream = static_cast<Http3Upstream *>(user_data);
2386
2387   if (upstream->http_reset_stream(stream_id, app_error_code) != 0) {
2388     return NGHTTP3_ERR_CALLBACK_FAILURE;
2389   }
2390
2391   return 0;
2392 }
2393 } // namespace
2394
2395 int Http3Upstream::http_reset_stream(int64_t stream_id,
2396                                      uint64_t app_error_code) {
2397   auto rv = ngtcp2_conn_shutdown_stream_write(conn_, stream_id, app_error_code);
2398   if (ngtcp2_err_is_fatal(rv)) {
2399     ULOG(ERROR, this) << "ngtcp2_conn_shutdown_stream_write: "
2400                       << ngtcp2_strerror(rv);
2401     return -1;
2402   }
2403
2404   return 0;
2405 }
2406
2407 int Http3Upstream::setup_httpconn() {
2408   int rv;
2409
2410   if (ngtcp2_conn_get_max_local_streams_uni(conn_) < 3) {
2411     return -1;
2412   }
2413
2414   nghttp3_callbacks callbacks{
2415       shrpx::http_acked_stream_data,
2416       shrpx::http_stream_close,
2417       shrpx::http_recv_data,
2418       http_deferred_consume,
2419       shrpx::http_begin_request_headers,
2420       shrpx::http_recv_request_header,
2421       shrpx::http_end_request_headers,
2422       nullptr, // begin_trailers
2423       nullptr, // recv_trailer
2424       nullptr, // end_trailers
2425       shrpx::http_send_stop_sending,
2426       shrpx::http_end_stream,
2427       shrpx::http_reset_stream,
2428   };
2429
2430   auto config = get_config();
2431
2432   nghttp3_settings settings;
2433   nghttp3_settings_default(&settings);
2434   settings.qpack_max_table_capacity = 4_k;
2435
2436   if (!config->http2_proxy) {
2437     settings.enable_connect_protocol = 1;
2438   }
2439
2440   auto mem = nghttp3_mem_default();
2441
2442   rv = nghttp3_conn_server_new(&httpconn_, &callbacks, &settings, mem, this);
2443   if (rv != 0) {
2444     ULOG(ERROR, this) << "nghttp3_conn_server_new: " << nghttp3_strerror(rv);
2445     return -1;
2446   }
2447
2448   ngtcp2_transport_params params;
2449   ngtcp2_conn_get_local_transport_params(conn_, &params);
2450
2451   nghttp3_conn_set_max_client_streams_bidi(httpconn_,
2452                                            params.initial_max_streams_bidi);
2453
2454   int64_t ctrl_stream_id;
2455
2456   rv = ngtcp2_conn_open_uni_stream(conn_, &ctrl_stream_id, nullptr);
2457   if (rv != 0) {
2458     ULOG(ERROR, this) << "ngtcp2_conn_open_uni_stream: " << ngtcp2_strerror(rv);
2459     return -1;
2460   }
2461
2462   rv = nghttp3_conn_bind_control_stream(httpconn_, ctrl_stream_id);
2463   if (rv != 0) {
2464     ULOG(ERROR, this) << "nghttp3_conn_bind_control_stream: "
2465                       << nghttp3_strerror(rv);
2466     return -1;
2467   }
2468
2469   int64_t qpack_enc_stream_id, qpack_dec_stream_id;
2470
2471   rv = ngtcp2_conn_open_uni_stream(conn_, &qpack_enc_stream_id, nullptr);
2472   if (rv != 0) {
2473     ULOG(ERROR, this) << "ngtcp2_conn_open_uni_stream: " << ngtcp2_strerror(rv);
2474     return -1;
2475   }
2476
2477   rv = ngtcp2_conn_open_uni_stream(conn_, &qpack_dec_stream_id, nullptr);
2478   if (rv != 0) {
2479     ULOG(ERROR, this) << "ngtcp2_conn_open_uni_stream: " << ngtcp2_strerror(rv);
2480     return -1;
2481   }
2482
2483   rv = nghttp3_conn_bind_qpack_streams(httpconn_, qpack_enc_stream_id,
2484                                        qpack_dec_stream_id);
2485   if (rv != 0) {
2486     ULOG(ERROR, this) << "nghttp3_conn_bind_qpack_streams: "
2487                       << nghttp3_strerror(rv);
2488     return -1;
2489   }
2490
2491   return 0;
2492 }
2493
2494 int Http3Upstream::error_reply(Downstream *downstream,
2495                                unsigned int status_code) {
2496   int rv;
2497   auto &resp = downstream->response();
2498
2499   auto &balloc = downstream->get_block_allocator();
2500
2501   auto html = http::create_error_html(balloc, status_code);
2502   resp.http_status = status_code;
2503   auto body = downstream->get_response_buf();
2504   body->append(html);
2505   downstream->set_response_state(DownstreamState::MSG_COMPLETE);
2506
2507   nghttp3_data_reader data_read;
2508   data_read.read_data = downstream_read_data_callback;
2509
2510   auto lgconf = log_config();
2511   lgconf->update_tstamp(std::chrono::system_clock::now());
2512
2513   auto response_status = http2::stringify_status(balloc, status_code);
2514   auto content_length = util::make_string_ref_uint(balloc, html.size());
2515   auto date = make_string_ref(balloc, lgconf->tstamp->time_http);
2516
2517   auto nva = std::array<nghttp3_nv, 5>{
2518       {http3::make_nv_ls_nocopy(":status", response_status),
2519        http3::make_nv_ll("content-type", "text/html; charset=UTF-8"),
2520        http3::make_nv_ls_nocopy("server", get_config()->http.server_name),
2521        http3::make_nv_ls_nocopy("content-length", content_length),
2522        http3::make_nv_ls_nocopy("date", date)}};
2523
2524   rv = nghttp3_conn_submit_response(httpconn_, downstream->get_stream_id(),
2525                                     nva.data(), nva.size(), &data_read);
2526   if (nghttp3_err_is_fatal(rv)) {
2527     ULOG(FATAL, this) << "nghttp3_conn_submit_response() failed: "
2528                       << nghttp3_strerror(rv);
2529     return -1;
2530   }
2531
2532   downstream->reset_upstream_wtimer();
2533
2534   if (shutdown_stream_read(downstream->get_stream_id(), NGHTTP3_H3_NO_ERROR) !=
2535       0) {
2536     return -1;
2537   }
2538
2539   return 0;
2540 }
2541
2542 int Http3Upstream::shutdown_stream(Downstream *downstream,
2543                                    uint64_t app_error_code) {
2544   auto stream_id = downstream->get_stream_id();
2545
2546   if (LOG_ENABLED(INFO)) {
2547     ULOG(INFO, this) << "Shutdown stream_id=" << stream_id
2548                      << " with app_error_code=" << app_error_code;
2549   }
2550
2551   auto rv = ngtcp2_conn_shutdown_stream(conn_, stream_id, app_error_code);
2552   if (rv != 0) {
2553     ULOG(FATAL, this) << "ngtcp2_conn_shutdown_stream() failed: "
2554                       << ngtcp2_strerror(rv);
2555     return -1;
2556   }
2557
2558   return 0;
2559 }
2560
2561 int Http3Upstream::shutdown_stream_read(int64_t stream_id,
2562                                         uint64_t app_error_code) {
2563   auto rv =
2564       ngtcp2_conn_shutdown_stream_read(conn_, stream_id, NGHTTP3_H3_NO_ERROR);
2565   if (ngtcp2_err_is_fatal(rv)) {
2566     ULOG(FATAL, this) << "ngtcp2_conn_shutdown_stream_read: "
2567                       << ngtcp2_strerror(rv);
2568     return -1;
2569   }
2570
2571   return 0;
2572 }
2573
2574 int Http3Upstream::redirect_to_https(Downstream *downstream) {
2575   auto &req = downstream->request();
2576   if (req.regular_connect_method() || req.scheme != "http") {
2577     return error_reply(downstream, 400);
2578   }
2579
2580   auto authority = util::extract_host(req.authority);
2581   if (authority.empty()) {
2582     return error_reply(downstream, 400);
2583   }
2584
2585   auto &balloc = downstream->get_block_allocator();
2586   auto config = get_config();
2587   auto &httpconf = config->http;
2588
2589   StringRef loc;
2590   if (httpconf.redirect_https_port == StringRef::from_lit("443")) {
2591     loc = concat_string_ref(balloc, StringRef::from_lit("https://"), authority,
2592                             req.path);
2593   } else {
2594     loc = concat_string_ref(balloc, StringRef::from_lit("https://"), authority,
2595                             StringRef::from_lit(":"),
2596                             httpconf.redirect_https_port, req.path);
2597   }
2598
2599   auto &resp = downstream->response();
2600   resp.http_status = 308;
2601   resp.fs.add_header_token(StringRef::from_lit("location"), loc, false,
2602                            http2::HD_LOCATION);
2603
2604   return send_reply(downstream, nullptr, 0);
2605 }
2606
2607 void Http3Upstream::consume(int64_t stream_id, size_t nconsumed) {
2608   ngtcp2_conn_extend_max_stream_offset(conn_, stream_id, nconsumed);
2609   ngtcp2_conn_extend_max_offset(conn_, nconsumed);
2610 }
2611
2612 void Http3Upstream::remove_downstream(Downstream *downstream) {
2613   if (downstream->accesslog_ready()) {
2614     handler_->write_accesslog(downstream);
2615   }
2616
2617   nghttp3_conn_set_stream_user_data(httpconn_, downstream->get_stream_id(),
2618                                     nullptr);
2619
2620   auto next_downstream = downstream_queue_.remove_and_get_blocked(downstream);
2621
2622   if (next_downstream) {
2623     initiate_downstream(next_downstream);
2624   }
2625
2626   if (downstream_queue_.get_downstreams() == nullptr) {
2627     // There is no downstream at the moment.  Start idle timer now.
2628     handler_->repeat_read_timer();
2629   }
2630 }
2631
2632 void Http3Upstream::log_response_headers(
2633     Downstream *downstream, const std::vector<nghttp3_nv> &nva) const {
2634   std::stringstream ss;
2635   for (auto &nv : nva) {
2636     ss << TTY_HTTP_HD << StringRef{nv.name, nv.namelen} << TTY_RST << ": "
2637        << StringRef{nv.value, nv.valuelen} << "\n";
2638   }
2639   ULOG(INFO, this) << "HTTP response headers. stream_id="
2640                    << downstream->get_stream_id() << "\n"
2641                    << ss.str();
2642 }
2643
2644 int Http3Upstream::check_shutdown() {
2645   auto worker = handler_->get_worker();
2646
2647   if (!worker->get_graceful_shutdown()) {
2648     return 0;
2649   }
2650
2651   ev_prepare_stop(handler_->get_loop(), &prep_);
2652
2653   return start_graceful_shutdown();
2654 }
2655
2656 int Http3Upstream::start_graceful_shutdown() {
2657   int rv;
2658
2659   if (ev_is_active(&shutdown_timer_)) {
2660     return 0;
2661   }
2662
2663   rv = nghttp3_conn_submit_shutdown_notice(httpconn_);
2664   if (rv != 0) {
2665     ULOG(FATAL, this) << "nghttp3_conn_submit_shutdown_notice: "
2666                       << nghttp3_strerror(rv);
2667     return -1;
2668   }
2669
2670   handler_->signal_write();
2671
2672   auto t = ngtcp2_conn_get_pto(conn_);
2673
2674   ev_timer_set(&shutdown_timer_, static_cast<ev_tstamp>(t * 3) / NGTCP2_SECONDS,
2675                0.);
2676   ev_timer_start(handler_->get_loop(), &shutdown_timer_);
2677
2678   return 0;
2679 }
2680
2681 int Http3Upstream::submit_goaway() {
2682   int rv;
2683
2684   rv = nghttp3_conn_shutdown(httpconn_);
2685   if (rv != 0) {
2686     ULOG(FATAL, this) << "nghttp3_conn_shutdown: " << nghttp3_strerror(rv);
2687     return -1;
2688   }
2689
2690   handler_->signal_write();
2691
2692   return 0;
2693 }
2694
2695 void Http3Upstream::idle_close() { idle_close_ = true; }
2696
2697 int Http3Upstream::open_qlog_file(const StringRef &dir,
2698                                   const ngtcp2_cid &scid) const {
2699   std::array<char, sizeof("20141115T125824.741+0900")> buf;
2700
2701   auto path = dir.str();
2702   path += '/';
2703   path +=
2704       util::format_iso8601_basic(buf.data(), std::chrono::system_clock::now());
2705   path += '-';
2706   path += util::format_hex(scid.data, scid.datalen);
2707   path += ".qlog";
2708
2709   int fd;
2710
2711 #ifdef O_CLOEXEC
2712   while ((fd = open(path.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC,
2713                     S_IRUSR | S_IWUSR | S_IRGRP)) == -1 &&
2714          errno == EINTR)
2715     ;
2716 #else  // !O_CLOEXEC
2717   while ((fd = open(path.c_str(), O_WRONLY | O_CREAT | O_TRUNC,
2718                     S_IRUSR | S_IWUSR | S_IRGRP)) == -1 &&
2719          errno == EINTR)
2720     ;
2721
2722   if (fd != -1) {
2723     util::make_socket_closeonexec(fd);
2724   }
2725 #endif // !O_CLOEXEC
2726
2727   if (fd == -1) {
2728     auto error = errno;
2729     ULOG(ERROR, this) << "Failed to open qlog file " << path
2730                       << ": errno=" << error;
2731     return -1;
2732   }
2733
2734   return fd;
2735 }
2736
2737 } // namespace shrpx