paivana

stream_upstream.c (35689B)

---

 /*
   This file is part of paivana tests.
   Copyright (C) 2026 Taler Systems SA
 
   Paivana is free software; you can redistribute it and/or
   modify it under the terms of the GNU Affero General Public License
   as published by the Free Software Foundation; either version
   3, or (at your option) any later version.
 
   Paivana is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty
   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
   the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public
   License along with Paivana; see the file COPYING.  If not,
   write to the Free Software Foundation, Inc., 51 Franklin
   Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
 
 /**
  * @file stream_upstream.c
  * @brief Raw-socket HTTP upstream for the streaming tests: serves
  *        bodies far larger than memory, at a controllable rate, and
  *        in the framings a conforming server library will not
  *        produce.
  *
  *        Raw sockets rather than libmicrohttpd because half of these
  *        targets are deliberate framing abuse -- a declared
  *        `Content-Length` that is not delivered, a chunked response
  *        with no terminating chunk -- which is exactly what a correct
  *        server refuses to emit.
  *
  *        Bodies are a deterministic function of the byte offset (see
  *        `pattern_at()'), so hundreds of megabytes can be verified
  *        without either side storing them.  The same function is in
  *        `stream_client.c'; the two must agree.
  *
  *        Targets, selected by request path:
  *
  *        /cl?bytes=N[&rate=R]       declare Content-Length: N, send N
  *        /chunked?bytes=N[&rate=R]  same body, chunked, no length
  *        /short?declare=N&send=M    declare N, send M < N, close
  *        /chunk-abort?after=N       chunked, N bytes, no terminator
  *        /hang?after=N              N bytes, then nothing, never close
  *        /slowstart?delay=D[&bytes=N]  headers now, first byte in D ms
  *        /sink[?rate=R]             read a request body, report its size
  *        /sink-early?after=N        answer 413 after N bytes of body
  *        /status?code=C[&len=N]     bare status, optional Content-Length
  *        /range?bytes=N             honour one Range, answer 206
  *        /mute                      accept, then never answer at all
  *        /trailers?bytes=N          chunked, with a trailer section
  *        /interim?bytes=N           a 103 before the final response
  *
  *        `rate' is in bytes per second; 0 (the default) means as fast
  *        as the socket takes it.  A rate limit is what makes the
  *        congestion cases reproducible: without it the kernel buffers
  *        absorb everything and no backpressure is ever exercised.
  */
 #ifndef _GNU_SOURCE
 #define _GNU_SOURCE
 #endif
 #include "platform.h"
 #include <gnunet/gnunet_util_lib.h>
 #include <limits.h>
 #include <poll.h>
 #include <strings.h>
 #include <time.h>
 
 /**
  * Bytes written per iteration of a rate-limited send loop.  Small
  * enough that a slow rate really does dribble rather than arriving in
  * one burst per second, large enough not to make a 200 MiB unlimited
  * transfer syscall-bound.
  */
 #define SEND_GRAIN 16384
 
 /**
  * How long (ms) `/hang' parks a connection before giving up on the
  * peer.  Only reached when paivana's own stall watchdog failed to
  * fire, which is the bug the case is looking for; the bound is here
  * so a wedged run still terminates.
  */
 #define HANG_TIMEOUT_MS 120000
 
 /**
  * Size of the request header block we are willing to read.
  */
 #define HDR_MAX 16384
 
 
 static volatile sig_atomic_t run_flag = 1;
 
 /**
  * Body bytes this connection moved, in whichever direction it was
  * moving them.  Reported when the connection is done; see `serve()'.
  */
 static uint64_t req_bytes;
 
 /**
  * What the connection was asked for, for the report line.  Sized to
  * hold a whole request target, so that a long query string is reported
  * rather than silently cut.
  */
 static char req_label[1024];
 
 
 static void
 on_sig (int sig)
 {
   (void) sig;
   run_flag = 0;
 }
 
 
 /**
  * The body byte at offset @a off.
  *
  * Deterministic and cheap, and deliberately not a constant: a body of
  * one repeated character would pass a comparison that dropped or
  * duplicated a whole aligned block, which is precisely the mistake a
  * ring buffer with wrong wrap arithmetic makes.  The `off >> 13' term
  * makes the pattern differ between 8 KiB blocks as well as within
  * them.
  *
  * Must agree with the same function in `stream_client.c'.
  *
  * @param off byte offset into the body
  * @return the byte that belongs there
  */
 static uint8_t
 pattern_at (uint64_t off)
 {
   return (uint8_t) ((off * 7) + (off >> 13));
 }
 
 
 /**
  * Write all of @a len bytes from @a buf to @a fd.
  *
  * @param fd socket to write to
  * @param buf bytes to write
  * @param len number of bytes in @a buf
  * @return true if all of it went out
  */
 static bool
 write_all (int fd,
            const void *buf,
            size_t len)
 {
   const char *p = buf;
   size_t off = 0;
 
   while (off < len)
   {
     ssize_t n = write (fd,
                        p + off,
                        len - off);
 
     if (n < 0)
     {
       if (EINTR == errno)
         continue;
       return false;
     }
     if (0 == n)
       return false;
     off += (size_t) n;
   }
   return true;
 }
 
 
 /**
  * Sleep for @a ms milliseconds, resuming across signals.
  */
 static void
 sleep_ms (unsigned int ms)
 {
   struct timespec ts = {
     .tv_sec = ms / 1000,
     .tv_nsec = (long) (ms % 1000) * 1000000L
   };
 
   while ( (0 != nanosleep (&ts, &ts)) &&
           (EINTR == errno) )
     ; /* again */
 }
 
 
 /**
  * Send @a total bytes of the pattern starting at offset @a start,
  * at most @a rate bytes per second.
  *
  * The rate is enforced by sleeping between grains rather than by
  * pacing against a clock, which would let a stalled peer be
  * compensated for with a burst -- and a burst is exactly what these
  * tests must not see.
  *
  * @param fd socket to write to
  * @param start first body offset to send
  * @param total number of bytes to send
  * @param rate bytes per second, 0 for unlimited
  * @return true if all of it went out
  */
 static bool
 send_pattern (int fd,
               uint64_t start,
               uint64_t total,
               unsigned long long rate)
 {
   static char grain[SEND_GRAIN];
   uint64_t sent = 0;
   size_t step = SEND_GRAIN;
   unsigned int nap = 0;
 
   if (0 != rate)
   {
     /* Aim for roughly ten writes a second, but never below one byte
        per write and never above a grain. */
     step = (size_t) GNUNET_MIN ((unsigned long long) SEND_GRAIN,
                                 GNUNET_MAX (1ULL, rate / 10));
     nap = (unsigned int) ((1000ULL * step) / rate);
   }
   while (sent < total)
   {
     size_t n = (size_t) GNUNET_MIN ((uint64_t) step,
                                     total - sent);
 
     for (size_t i = 0; i < n; i++)
       grain[i] = (char) pattern_at (start + sent + i);
     if (! write_all (fd,
                      grain,
                      n))
       return false;
     sent += n;
     if ( (0 != nap) &&
          (sent < total) )
       sleep_ms (nap);
     if (! run_flag)
       return false;
   }
   return true;
 }
 
 
 /**
  * Like `send_pattern()', but reports how much actually went out rather
  * than only whether all of it did.
  *
  * The count is the backpressure evidence.  A proxy that assembled the
  * body in memory would take everything at line rate however slowly its
  * own client was reading; one that relays it can only take what the
  * client has made room for.  So how far this gets, and how long it
  * takes to get there, is what says which of the two is happening --
  * and it cannot be seen from the client end, where both look the same.
  *
  * @param fd socket to write to
  * @param start first body offset to send
  * @param total number of bytes to send
  * @param rate bytes per second, 0 for unlimited
  * @return number of bytes written
  */
 static uint64_t
 send_pattern_counted (int fd,
                       uint64_t start,
                       uint64_t total,
                       unsigned long long rate)
 {
   static char grain[SEND_GRAIN];
   uint64_t sent = 0;
   size_t step = SEND_GRAIN;
   unsigned int nap = 0;
 
   if (0 != rate)
   {
     step = (size_t) GNUNET_MIN ((unsigned long long) SEND_GRAIN,
                                 GNUNET_MAX (1ULL, rate / 10));
     nap = (unsigned int) ((1000ULL * step) / rate);
   }
   while (sent < total)
   {
     size_t n = (size_t) GNUNET_MIN ((uint64_t) step,
                                     total - sent);
 
     for (size_t i = 0; i < n; i++)
       grain[i] = (char) pattern_at (start + sent + i);
     if (! write_all (fd,
                      grain,
                      n))
       return sent;
     sent += n;
     if ( (0 != nap) &&
          (sent < total) )
       sleep_ms (nap);
     if (! run_flag)
       return sent;
   }
   return sent;
 }
 
 
 /**
  * Body bytes the last `send_chunked()' managed to write, which is not
  * the same as what it was asked for when the peer stops reading.
  */
 static uint64_t chunked_sent;
 
 
 /**
  * Send @a total bytes of the pattern as HTTP chunks.
  *
  * @param fd socket to write to
  * @param total number of body bytes
  * @param rate bytes per second, 0 for unlimited
  * @param terminate write the terminating zero-length chunk
  * @return true if all of it went out
  */
 static bool
 send_chunked (int fd,
               uint64_t total,
               unsigned long long rate,
               bool terminate)
 {
   static char grain[SEND_GRAIN];
   uint64_t sent = 0;
   size_t step = SEND_GRAIN;
   unsigned int nap = 0;
   char hdr[32];
 
   chunked_sent = 0;
 
   if (0 != rate)
   {
     step = (size_t) GNUNET_MIN ((unsigned long long) SEND_GRAIN,
                                 GNUNET_MAX (1ULL, rate / 10));
     nap = (unsigned int) ((1000ULL * step) / rate);
   }
   while (sent < total)
   {
     size_t n = (size_t) GNUNET_MIN ((uint64_t) step,
                                     total - sent);
     int hl = snprintf (hdr,
                        sizeof (hdr),
                        "%zx\r\n",
                        n);
 
     for (size_t i = 0; i < n; i++)
       grain[i] = (char) pattern_at (sent + i);
     if ( (! write_all (fd, hdr, (size_t) hl)) ||
          (! write_all (fd, grain, n)) ||
          (! write_all (fd, "\r\n", 2)) )
       return false;
     sent += n;
     chunked_sent = sent;
     if ( (0 != nap) &&
          (sent < total) )
       sleep_ms (nap);
     if (! run_flag)
       return false;
   }
   if (terminate)
     return write_all (fd,
                       "0\r\n\r\n",
                       5);
   return true;
 }
 
 
 /**
  * Read from @a fd into @a buf until "\r\n\r\n" appears.
  *
  * @param fd socket to read from
  * @param[out] buf where to put the header block
  * @param cap capacity of @a buf
  * @param[out] eoh set to the offset just past the CRLFCRLF
  * @return total bytes read, or -1 on error or overlong headers
  */
 static ssize_t
 read_until_eoh (int fd,
                 char *buf,
                 size_t cap,
                 size_t *eoh)
 {
   size_t pos = 0;
   size_t scan = 0;
 
   while (pos < cap)
   {
     ssize_t n = read (fd,
                       buf + pos,
                       cap - pos);
 
     if (n <= 0)
       return -1;
     pos += (size_t) n;
     while (scan + 3 < pos)
     {
       if ( ('\r' == buf[scan]) &&
            ('\n' == buf[scan + 1]) &&
            ('\r' == buf[scan + 2]) &&
            ('\n' == buf[scan + 3]) )
       {
         *eoh = scan + 4;
         return (ssize_t) pos;
       }
       scan++;
     }
   }
   return -1;
 }
 
 
 /**
  * Value of the query parameter @a name in the request target @a
  * target, or @a dflt if it is absent or unparseable.
  *
  * @param target request target, NUL-terminated
  * @param name parameter name
  * @param dflt value to use if it is not there
  * @return the parsed value
  */
 static unsigned long long
 query_num (const char *target,
            const char *name,
            unsigned long long dflt)
 {
   const char *q = strchr (target, '?');
   size_t nlen = strlen (name);
 
   if (NULL == q)
     return dflt;
   for (const char *p = q + 1; '\0' != *p;)
   {
     if ( (0 == strncmp (p, name, nlen)) &&
          ('=' == p[nlen]) )
     {
       char *end;
       unsigned long long v;
 
       errno = 0;
       v = strtoull (&p[nlen + 1],
                     &end,
                     10);
       if ( (0 != errno) ||
            (end == &p[nlen + 1]) )
         return dflt;
       return v;
     }
     p = strchr (p, '&');
     if (NULL == p)
       break;
     p++;
   }
   return dflt;
 }
 
 
 /**
  * Does @a target name the path @a path (ignoring any query)?
  */
 static bool
 path_is (const char *target,
          const char *path)
 {
   size_t plen = strlen (path);
 
   return (0 == strncmp (target, path, plen)) &&
          ( ('\0' == target[plen]) ||
            ('?' == target[plen]) );
 }
 
 
 /**
  * Find the Content-Length in the request header block.
  *
  * @param hdr start of the (not NUL-terminated) header block
  * @param len number of bytes in @a hdr
  * @return the announced length, or -1 if there was none
  */
 static long long
 find_content_length (const char *hdr,
                      size_t len)
 {
   static const char name[] = "content-length:";
   const size_t nlen = sizeof (name) - 1;
 
   for (size_t i = 0; i + nlen <= len; i++)
   {
     size_t j;
     long long v = 0;
     bool digits = false;
 
     if ( (0 == i) ||
          ('\n' != hdr[i - 1]) )
       continue;
     if (0 != strncasecmp (&hdr[i],
                           name,
                           nlen))
       continue;
     j = i + nlen;
     while ( (j < len) &&
             ( (' ' == hdr[j]) || ('\t' == hdr[j]) ) )
       j++;
     while ( (j < len) &&
             ('0' <= hdr[j]) &&
             ('9' >= hdr[j]) )
     {
       if (v > (LLONG_MAX - (hdr[j] - '0')) / 10)
         return -1;
       v = v * 10 + (hdr[j] - '0');
       digits = true;
       j++;
     }
     return digits ? v : -1;
   }
   return -1;
 }
 
 
 /**
  * Is the request header block chunked?
  *
  * @param hdr start of the header block
  * @param len number of bytes in @a hdr
  * @return true if a Transfer-Encoding naming chunked is present
  */
 static bool
 is_chunked_request (const char *hdr,
                     size_t len)
 {
   static const char name[] = "transfer-encoding:";
   const size_t nlen = sizeof (name) - 1;
 
   for (size_t i = 0; i + nlen <= len; i++)
   {
     if ( (0 == i) ||
          ('\n' != hdr[i - 1]) )
       continue;
     if (0 == strncasecmp (&hdr[i],
                           name,
                           nlen))
       return true;
   }
   return false;
 }
 
 
 /**
  * Read and verify a request body, at most @a rate bytes per second.
  *
  * Verification is against `pattern_at()', so a body that arrived
  * complete but reordered or with a duplicated block is caught -- the
  * byte count alone would not notice.  Both framings are handled: a
  * declared length, and chunked, which is what libcurl sends when the
  * length is not known in advance.
  *
  * @param fd socket to read from
  * @param pre bytes of body already read with the headers
  * @param pre_len number of bytes in @a pre
  * @param declared Content-Length, or -1 for chunked/unknown
  * @param chunked read the body as chunks
  * @param rate bytes per second, 0 for unlimited
  * @param stop_after stop reading after this many body bytes, 0 for
  *        "read it all"
  * @param[out] ok set to false if the pattern did not match
  * @return number of body bytes read
  */
 static uint64_t
 drain_body (int fd,
             const char *pre,
             size_t pre_len,
             long long declared,
             bool chunked,
             unsigned long long rate,
             uint64_t stop_after,
             bool *ok)
 {
   static char buf[SEND_GRAIN];
   uint64_t got = 0;
   size_t step = sizeof (buf);
   unsigned int nap = 0;
   /* Chunk framing is stripped by a tiny state machine rather than a
      parser: we only ever read what libcurl writes.  The state has to
      survive a read() boundary falling anywhere -- including between a
      chunk's last data byte and its trailing CRLF, which is why
      skipping that CRLF is a state and not two bytes consumed on the
      spot.  Getting that wrong makes the next read start on "\r\n",
      which parses as an empty chunk-size line, which reads as the
      terminating chunk: the body ends early and the byte count is
      short but self-consistent. */
   enum
   {
     CH_HDR,   /* accumulating the chunk-size line */
     CH_DATA,  /* @e chunk_left bytes of chunk data to come */
     CH_CRLF   /* @e crlf_left bytes of the post-data CRLF to skip */
   } cstate = CH_HDR;
   uint64_t chunk_left = 0;
   unsigned int crlf_left = 0;
   char hdrline[32];
   size_t hdrpos = 0;
 
   *ok = true;
   if (0 != rate)
   {
     step = (size_t) GNUNET_MIN ((unsigned long long) sizeof (buf),
                                 GNUNET_MAX (1ULL, rate / 10));
     nap = (unsigned int) ((1000ULL * step) / rate);
   }
   while (run_flag)
   {
     const char *p;
     size_t avail;
     ssize_t n;
 
     if (0 != pre_len)
     {
       p = pre;
       avail = pre_len;
       pre_len = 0;
     }
     else
     {
       if ( (! chunked) &&
            (declared >= 0) &&
            (got >= (uint64_t) declared) )
         break;
       if ( (0 != stop_after) &&
            (got >= stop_after) )
         break;
       n = read (fd,
                 buf,
                 step);
       if (n <= 0)
         break;
       p = buf;
       avail = (size_t) n;
       if (0 != nap)
         sleep_ms (nap);
     }
     while (0 != avail)
     {
       if (! chunked)
       {
         for (size_t i = 0; i < avail; i++)
           if ((uint8_t) p[i] != pattern_at (got + i))
           {
             *ok = false;
             break;
           }
         got += avail;
         avail = 0;
         continue;
       }
       switch (cstate)
       {
       case CH_HDR:
         /* Accumulate up to the CRLF that ends the chunk size line. */
         while ( (0 != avail) &&
                 (hdrpos + 1 < sizeof (hdrline)) )
         {
           char c = *p++;
 
           avail--;
           hdrline[hdrpos++] = c;
           if ( (hdrpos >= 2) &&
                ('\r' == hdrline[hdrpos - 2]) &&
                ('\n' == hdrline[hdrpos - 1]) )
           {
             hdrline[hdrpos - 2] = '\0';
             chunk_left = strtoull (hdrline,
                                    NULL,
                                    16);
             hdrpos = 0;
             cstate = (0 == chunk_left) ? CH_HDR : CH_DATA;
             break;
           }
         }
         if ( (CH_HDR == cstate) &&
              (0 == hdrpos) )
           return got; /* terminating chunk */
         break;
       case CH_DATA:
         {
           size_t take = (size_t) GNUNET_MIN ((uint64_t) avail,
                                              chunk_left);
 
           for (size_t i = 0; i < take; i++)
             if ((uint8_t) p[i] != pattern_at (got + i))
             {
               *ok = false;
               break;
             }
           got += take;
           p += take;
           avail -= take;
           chunk_left -= take;
           if (0 == chunk_left)
           {
             cstate = CH_CRLF;
             crlf_left = 2;
           }
         }
         break;
       case CH_CRLF:
         {
           size_t skip = (size_t) GNUNET_MIN ((uint64_t) avail,
                                              (uint64_t) crlf_left);
 
           p += skip;
           avail -= skip;
           crlf_left -= (unsigned int) skip;
           if (0 == crlf_left)
           {
             cstate = CH_HDR;
             hdrpos = 0;
           }
         }
         break;
       }
     }
   }
   return got;
 }
 
 
 /**
  * Serve one connection.
  *
  * @param fd the accepted socket
  */
 static void
 serve (int fd)
 {
   char hdr[HDR_MAX];
   char target[1024];
   size_t eoh = 0;
   ssize_t total;
   char resp[512];
   int rl;
   unsigned long long rate;
 
   total = read_until_eoh (fd,
                           hdr,
                           sizeof (hdr),
                           &eoh);
   if (0 > total)
     return;
   {
     /* Request line: METHOD SP target SP version */
     const char *sp1 = memchr (hdr, ' ', (size_t) total);
     const char *sp2;
     size_t tlen;
 
     if (NULL == sp1)
       return;
     sp2 = memchr (sp1 + 1,
                   ' ',
                   (size_t) total - (size_t) (sp1 + 1 - hdr));
     if (NULL == sp2)
       return;
     tlen = (size_t) (sp2 - sp1 - 1);
     if (tlen >= sizeof (target))
       return;
     memcpy (target,
             sp1 + 1,
             tlen);
     target[tlen] = '\0';
   }
   rate = query_num (target,
                     "rate",
                     0);
   snprintf (req_label,
             sizeof (req_label),
             "%s",
             target);
 
   if (path_is (target, "/cl"))
   {
     uint64_t bytes = query_num (target,
                                 "bytes",
                                 1024);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Content-Length: %llu\r\n"
                    "\r\n",
                    (unsigned long long) bytes);
     if (write_all (fd, resp, (size_t) rl))
       req_bytes = send_pattern_counted (fd,
                                         0,
                                         bytes,
                                         rate);
     return;
   }
   if (path_is (target, "/chunked"))
   {
     uint64_t bytes = query_num (target,
                                 "bytes",
                                 1024);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n");
     if (write_all (fd, resp, (size_t) rl))
       (void) send_chunked (fd,
                            bytes,
                            rate,
                            true);
     req_bytes = chunked_sent;
     return;
   }
   if (path_is (target, "/short"))
   {
     uint64_t declare = query_num (target,
                                   "declare",
                                   100000);
     uint64_t send = query_num (target,
                                "send",
                                1000);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Content-Length: %llu\r\n"
                    "\r\n",
                    (unsigned long long) declare);
     if (write_all (fd, resp, (size_t) rl))
       (void) send_pattern (fd,
                            0,
                            send,
                            rate);
     return;
   }
   if (path_is (target, "/chunk-abort"))
   {
     uint64_t after = query_num (target,
                                 "after",
                                 1000);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n");
     if (write_all (fd, resp, (size_t) rl))
       (void) send_chunked (fd,
                            after,
                            rate,
                            false); /* no terminating chunk */
     return;
   }
   if (path_is (target, "/hang"))
   {
     uint64_t after = query_num (target,
                                 "after",
                                 1000);
     struct pollfd pfd = {
       .fd = fd,
       .events = POLLRDHUP
     };
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n");
     if (! write_all (fd, resp, (size_t) rl))
       return;
     if (! send_chunked (fd,
                         after,
                         rate,
                         false))
       return;
     /* Now say nothing, and do not close: the point is a connection
        that is open, established and idle, which only paivana's own
        stall watchdog can resolve. */
     (void) poll (&pfd,
                  1,
                  HANG_TIMEOUT_MS);
     return;
   }
   if (path_is (target, "/slowstart"))
   {
     uint64_t delay = query_num (target,
                                 "delay",
                                 1000);
     uint64_t bytes = query_num (target,
                                 "bytes",
                                 1024);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Content-Length: %llu\r\n"
                    "\r\n",
                    (unsigned long long) bytes);
     if (! write_all (fd, resp, (size_t) rl))
       return;
     sleep_ms ((unsigned int) delay);
     (void) send_pattern (fd,
                          0,
                          bytes,
                          rate);
     return;
   }
   if (path_is (target, "/sink"))
   {
     long long declared = find_content_length (hdr,
                                               eoh);
     bool chunked = is_chunked_request (hdr,
                                        eoh);
     bool ok;
     uint64_t got;
     char body[128];
     int bl;
 
     got = drain_body (fd,
                       &hdr[eoh],
                       (size_t) total - eoh,
                       declared,
                       chunked,
                       rate,
                       0,
                       &ok);
     req_bytes = got;
     bl = snprintf (body,
                    sizeof (body),
                    "bytes=%llu framing=%s pattern=%s\n",
                    (unsigned long long) got,
                    chunked ? "chunked" : "length",
                    ok ? "ok" : "CORRUPT");
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: text/plain\r\n"
                    "Content-Length: %d\r\n"
                    "\r\n",
                    bl);
     if (write_all (fd, resp, (size_t) rl))
       (void) write_all (fd, body, (size_t) bl);
     return;
   }
   if (path_is (target, "/sink-early"))
   {
     uint64_t after = query_num (target,
                                 "after",
                                 1024);
     long long declared = find_content_length (hdr,
                                               eoh);
     bool chunked = is_chunked_request (hdr,
                                        eoh);
     bool ok;
     static const char body[] = "too large\n";
 
     (void) drain_body (fd,
                        &hdr[eoh],
                        (size_t) total - eoh,
                        declared,
                        chunked,
                        rate,
                        after,
                        &ok);
     /* Answer while the body is (very probably) still coming.  This is
        the case that only exists because the request is streamed: with
        the body buffered first, the origin could not have answered
        before seeing all of it. */
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 413 Content Too Large\r\n"
                    "Content-Type: text/plain\r\n"
                    "Connection: close\r\n"
                    "Content-Length: %zu\r\n"
                    "\r\n%s",
                    sizeof (body) - 1,
                    body);
     (void) write_all (fd, resp, (size_t) rl);
     return;
   }
   if (path_is (target, "/range"))
   {
     /* Honour a single "bytes=A-B" range, so that the 206 and its
        Content-Range can be checked end to end.  Nothing here parses
        the general grammar: the driver sends one form. */
     uint64_t bytes = query_num (target,
                                 "bytes",
                                 1048576);
     unsigned long long from = 0;
     unsigned long long to = bytes - 1;
     const char *r = NULL;
 
     for (size_t i = 0; i + 6 <= eoh; i++)
       if ( (0 != i) &&
            ('\n' == hdr[i - 1]) &&
            (0 == strncasecmp (&hdr[i], "range:", 6)) )
       {
         r = &hdr[i + 6];
         break;
       }
     if (NULL != r)
     {
       const char *eq = strchr (r, '=');
 
       if (NULL != eq)
       {
         char *end;
 
         from = strtoull (eq + 1, &end, 10);
         if ('-' == *end)
         {
           unsigned long long t = strtoull (end + 1, &end, 10);
 
           if (0 != t)
             to = t;
         }
       }
     }
     if (to >= bytes)
       to = bytes - 1;
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 206 Partial Content\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Content-Range: bytes %llu-%llu/%llu\r\n"
                    "Content-Length: %llu\r\n"
                    "\r\n",
                    from,
                    to,
                    (unsigned long long) bytes,
                    to - from + 1);
     if (write_all (fd, resp, (size_t) rl))
       (void) send_pattern (fd,
                            from,
                            to - from + 1,
                            rate);
     return;
   }
   if (path_is (target, "/mute"))
   {
     /* Accept the connection and say nothing at all.  Distinct from an
        origin that is not there, which is a 502: this one is a 504, and
        telling them apart is what paivana's time-to-headers clock is
        for. */
     struct pollfd pfd = {
       .fd = fd,
       .events = POLLRDHUP
     };
 
     (void) poll (&pfd,
                  1,
                  HANG_TIMEOUT_MS);
     return;
   }
   if (path_is (target, "/trailers"))
   {
     /* A chunked response with a trailer section.  RFC 9110 §6.5.1
        forbids merging a trailer into the header section, and the
        response has in any case already been queued by the time these
        arrive, so they must be dropped rather than added to it. */
     uint64_t bytes = query_num (target,
                                 "bytes",
                                 4096);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Trailer: X-Trailer-Check\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n");
     if (! write_all (fd, resp, (size_t) rl))
       return;
     if (! send_chunked (fd,
                         bytes,
                         rate,
                         false))
       return;
     (void) write_all (fd,
                       "0\r\n"
                       "X-Trailer-Check: leaked\r\n"
                       "\r\n",
                       strlen ("0\r\nX-Trailer-Check: leaked\r\n\r\n"));
     return;
   }
   if (path_is (target, "/interim"))
   {
     /* A 1xx before the final response, carrying a header that must not
        reappear on it. */
     uint64_t bytes = query_num (target,
                                 "bytes",
                                 4096);
 
     rl = snprintf (resp,
                    sizeof (resp),
                    "HTTP/1.1 103 Early Hints\r\n"
                    "X-Interim-Check: leaked\r\n"
                    "\r\n"
                    "HTTP/1.1 200 OK\r\n"
                    "Content-Type: application/octet-stream\r\n"
                    "Content-Length: %llu\r\n"
                    "\r\n",
                    (unsigned long long) bytes);
     if (write_all (fd, resp, (size_t) rl))
       (void) send_pattern (fd,
                            0,
                            bytes,
                            rate);
     return;
   }
   if (path_is (target, "/status"))
   {
     unsigned long long code = query_num (target,
                                          "code",
                                          204);
     unsigned long long len = query_num (target,
                                         "len",
                                         UINT64_MAX);
 
     if (UINT64_MAX == len)
       rl = snprintf (resp,
                      sizeof (resp),
                      "HTTP/1.1 %llu Status\r\n"
                      "\r\n",
                      code);
     else
       rl = snprintf (resp,
                      sizeof (resp),
                      "HTTP/1.1 %llu Status\r\n"
                      "Content-Length: %llu\r\n"
                      "\r\n",
                      code,
                      len);
     (void) write_all (fd, resp, (size_t) rl);
     return;
   }
   rl = snprintf (resp,
                  sizeof (resp),
                  "HTTP/1.1 404 Not Found\r\n"
                  "Content-Length: 0\r\n"
                  "\r\n");
   (void) write_all (fd, resp, (size_t) rl);
 }
 
 
 int
 main (int argc,
       char **argv)
 {
   int lsock;
   int port;
   int one = 1;
   struct sockaddr_in addr;
 
   if (2 > argc)
   {
     fprintf (stderr,
              "usage: %s PORT\n",
              argv[0]);
     return 1;
   }
   port = atoi (argv[1]);
   signal (SIGINT, &on_sig);
   signal (SIGTERM, &on_sig);
   /* A peer that goes away mid-body is the normal case here, not an
      error; without this the first such write kills the process. */
   signal (SIGPIPE, SIG_IGN);
   lsock = socket (AF_INET,
                   SOCK_STREAM,
                   0);
   if (0 > lsock)
   {
     perror ("socket");
     return 1;
   }
   (void) setsockopt (lsock,
                      SOL_SOCKET,
                      SO_REUSEADDR,
                      &one,
                      sizeof (one));
   memset (&addr, 0, sizeof (addr));
   addr.sin_family = AF_INET;
   addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
   addr.sin_port = htons ((uint16_t) port);
   if (0 != bind (lsock,
                  (struct sockaddr *) &addr,
                  sizeof (addr)))
   {
     perror ("bind");
     return 1;
   }
   if (0 != listen (lsock, 64))
   {
     perror ("listen");
     return 1;
   }
   fprintf (stderr,
            "stream_upstream listening on port %d\n",
            port);
   fflush (stderr);
   while (run_flag)
   {
     int fd = accept (lsock, NULL, NULL);
     pid_t pid;
 
     if (0 > fd)
     {
       if (EINTR == errno)
         continue;
       break;
     }
     /* One process per connection.  Not for throughput but because
        several of the targets deliberately do not finish -- `/hang'
        parks for two minutes, `/cl?rate=' dribbles for as long as the
        driver asks -- and a single-threaded accept loop would make the
        concurrency cases serialise behind them. */
     pid = fork ();
     if (0 == pid)
     {
       struct timespec t0;
       struct timespec t1;
 
       close (lsock);
       clock_gettime (CLOCK_MONOTONIC,
                      &t0);
       serve (fd);
       clock_gettime (CLOCK_MONOTONIC,
                      &t1);
       /* One line per connection, for the cases that assert on how the
          origin was paced rather than on what the client received.
          Written after the socket work so the timing covers it, and to
          stderr so it lands in the driver's log unbuffered. */
       fprintf (stderr,
                "served target=%s bytes=%llu ms=%ld\n",
                req_label,
                (unsigned long long) req_bytes,
                (long) ((t1.tv_sec - t0.tv_sec) * 1000
                        + (t1.tv_nsec - t0.tv_nsec) / 1000000));
       fflush (stderr);
       close (fd);
       _exit (0);
     }
     close (fd);
     if (0 > pid)
       perror ("fork");
     /* Reap whatever has finished; no zombies, no blocking. */
     while (0 < waitpid (-1, NULL, WNOHANG))
       ; /* again */
   }
   close (lsock);
   return 0;
 }
 
 
 /* end of stream_upstream.c */