commit 7e019a138d550ce866b19c370b00f35fb4988ac7
parent 9ec86ab5a3b3d90aca53c39bd75497f1ad0f933e
Author: Christian Grothoff <christian@grothoff.org>
Date: Fri, 7 Aug 2026 20:25:33 +0200
add reverse proxy test
Diffstat:
18 files changed, 4855 insertions(+), 738 deletions(-)
diff --git a/NEWS b/NEWS
@@ -0,0 +1,46 @@
+Unreleased:
+ - The reverse proxy streams both directions. A request or response
+ body is relayed as it arrives instead of being assembled in memory
+ first, so its size is no longer bounded by memory -- previously
+ anything over 40 MiB was refused outright with 502, and the client
+ could not receive a single byte until the origin had sent the last
+ one. The origin's framing now reaches the client unchanged as
+ well: a declared Content-Length stays declared and a chunked
+ response stays chunked, where before both were recomputed from the
+ assembled buffer. A HEAD accordingly reports the length the
+ equivalent GET would have had, rather than 0.
+
+ - Configuration: REQUEST_BUFFER_MAX (and its -u/--max-upload flag)
+ now sizes the buffer a request body is relayed through, 256 KiB by
+ default, and no longer decides how large an upload may be. That is
+ MAX_REQUEST_SIZE, new, 1 MiB by default. RESPONSE_BUFFER_MAX is
+ the mirror of the first; there is deliberately no mirror of the
+ second, since removing the limit on a response is the point of the
+ change.
+
+ UPGRADING: a configuration that set REQUEST_BUFFER_MAX (or -u) to
+ permit large uploads meant the limit, so where it is set and
+ MAX_REQUEST_SIZE is not, MAX_REQUEST_SIZE takes its value. Such a
+ configuration therefore keeps behaving as it did, at the cost of a
+ correspondingly large relay buffer; set both explicitly to separate
+ them.
+
+ - Timeouts: UPSTREAM_TIMEOUT (60 s) now bounds the time to the
+ response *headers* rather than to the whole transfer, since a large
+ download legitimately runs for longer than any useful ceiling. A
+ stalled transfer is bounded instead by the new
+ UPSTREAM_STALL_TIMEOUT (60 s), which is how long the origin may
+ move no bytes in either direction, and which does not run while
+ paivana is holding the origin back because the client has not
+ caught up.
+
+ - An origin that fails after its response headers have gone out can
+ no longer be reported as 502: the status is already on the wire.
+ It is reported as a framing error instead -- a declared
+ Content-Length that is not met, or a chunked response closed
+ without its terminating chunk -- which RFC 9112 section 8.1.2
+ requires a client to treat as a failure. The exception is an
+ HTTP/1.0 client receiving a response of unknown length, where the
+ close is the framing and truncation cannot be signalled.
+
+ - Requires GNUnet 0.29.0 for GNUNET_CURL_job_add_stream().
diff --git a/README b/README
@@ -41,8 +41,9 @@ How it works
two must agree — bounded above by the contract's `max_pickup_time`,
which is what the merchant's `max_pickup_delay` on the template sets.
7. Requests with a valid cookie are forwarded to the upstream server
- via libcurl. Neither direction is streamed: the body is buffered in
- full before it is passed on (see "Architecture notes").
+ via libcurl. Both directions are streamed: bytes are passed on as
+ they arrive, so the size of a proxied body is not bounded by memory
+ (see "Architecture notes").
The cookie carries a keyed hash over `(expiration time, website, client
address)` keyed by a `paivana_secret` derived from the configured
@@ -196,10 +197,12 @@ Besides the options GNUnet gives every program (`-c` / `--config`,
headers. Only safe behind a proxy that overwrites
them; see "Deployment behind a reverse proxy".
-u, --max-upload BYTES
- Request body to buffer before forwarding, default
- 1048576. Raising it past the 40 MiB the
- implementation caps every body at has no effect;
- anything larger is refused either way.
+ Bytes of a request body held in memory at once
+ while relaying it upstream, default 262144. A
+ throughput knob, not a limit: the largest body
+ accepted is MAX_REQUEST_SIZE. For configurations
+ written when these were one number, setting this
+ and not MAX_REQUEST_SIZE still sets both.
The daemon does not serve requests until it has fetched paywall templates
from the merchant backend. If template loading fails, startup is aborted:
@@ -419,33 +422,59 @@ Single-threaded event loop: GNUnet scheduler drives both inbound HTTP
Running multiple `paivana-httpd` processes on the same port is
supported as the main way to scale-up the system.
-Requests and responses are currently not streamed in the reverse
-proxy; as a result, uploads and downloads are currently hard-capped by
-the implementation at 40 MiB. Lower limits for the upload may be
-configured.
+Requests and responses are streamed in both directions: each is moved
+through a fixed-size ring buffer (REQUEST_BUFFER_MAX and
+RESPONSE_BUFFER_MAX, 256 KiB each by default) rather than assembled
+whole, so the size of a proxied body is bounded by nothing in Paivana.
+When the client cannot keep up, Paivana stops reading from the origin;
+when the origin cannot keep up, it stops reading from the client. The
+memory an in-flight request costs is therefore the two buffers, and the
+worst case is that times CONNECTION_LIMIT.
+
+An upload is still bounded, by MAX_REQUEST_SIZE (1 MiB by default),
+because accepting one is a policy decision rather than a memory
+constraint. A response is not bounded at all: an operator who wants to
+bound what their origin serves can do it at the origin.
The MHD daemon is not started until paywall templates have been fetched
from the merchant backend asynchronously.
-Because the whole response is buffered before anything is sent on, a
-1xx interim response cannot be forwarded — RFC 9110 §15.2 asks a proxy
-to forward them, and Paivana instead drops them. `103 Early Hints`
+A 1xx interim response is not forwarded — RFC 9110 §15.2 asks a proxy
+to forward them, and Paivana instead drops them, because the response
+MHD is handed is a single final one. `103 Early Hints`
therefore does not reach clients through Paivana. Its header fields
are dropped with it rather than being merged into the final response,
which is the part that would be actively harmful. Trailer fields are
dropped for the same reason (RFC 9110 §6.5.1 forbids merging them into
the header section).
-An origin that accepts the connection but does not finish a response
-within 60 seconds yields `504 Gateway Timeout`; one that cannot be
-reached at all yields `502 Bad Gateway`. The distinction matters
-because caches and monitoring retry the former and not the latter.
-
-Those 60 seconds are fixed and not configurable, and they are the
-effective ceiling on a whole request. The MHD connection timeout does
-not bound the phase spent waiting on the origin: the connection is
-suspended for its duration, and MHD drops suspended connections from
-its timeout lists.
+An origin that accepts the connection but does not produce response
+*headers* within `UPSTREAM_TIMEOUT` (60 s) yields `504 Gateway
+Timeout`; one that cannot be reached at all yields `502 Bad Gateway`.
+The distinction matters because caches and monitoring retry the former
+and not the latter. That clock is cancelled once the header section
+ends, and it is the only one that can still produce a status code:
+after it, the status is already on the wire.
+
+There is deliberately no ceiling on how long a request may take — a
+large download legitimately runs for as long as it runs. What is
+bounded instead is a *stall*: `UPSTREAM_STALL_TIMEOUT` (60 s) is how
+long the origin may move no bytes in either direction. The clock does
+not run while Paivana is itself holding the origin back because the
+client has not drained what has already arrived, so a client on a slow
+link is never mistaken for a slow origin.
+
+The MHD connection timeout does not cover any of this: a connection
+waiting on the origin is suspended, and MHD drops suspended connections
+from its timeout lists.
+
+Once the response headers have gone out the status cannot be retracted,
+so an origin that fails mid-body can only be reported as a framing
+error: a declared `Content-Length` that is not met, or a chunked
+response closed without its terminating chunk. Both are required to be
+treated as failures by RFC 9112 §8.1.2. The exception is an HTTP/1.0
+client receiving a response of unknown length, where the close *is* the
+framing and truncation is indistinguishable from success.
`OPTIONS` carrying `Max-Forwards: 0` is answered by Paivana itself with
an `Allow` list, as RFC 9110 §7.6.2 requires of an intermediary; any
diff --git a/configure b/configure
@@ -2,7 +2,7 @@
pkg_name="paivana"
pkg_default_features="doc logging"
-pkg_optional_features="coverage only-doc install-rpath"
+pkg_optional_features="coverage only-doc install-rpath sanitizers"
pkg_optional_dependencies=""
# DO NOT EDIT BELOW THIS LINE
@@ -215,6 +215,23 @@ for feat in $pkg_optional_features $pkg_default_features; do
false) mval=no ;;
esac
;;
+ sanitizers)
+ # meson's sanitizer switch is build-wide, so this is a
+ # build variant rather than an extra test. Never enabled
+ # for a package build: ASan is not available on every
+ # architecture Debian builds for, it adds a libasan
+ # dependency, and a leak in a dependency the suppression
+ # file has not learned about would fail the build for a
+ # reason that is not paivana's bug. The ownership
+ # handshake it protects is identical everywhere, so
+ # building it once in CI finds what building it eleven
+ # times would.
+ mfeat="b_sanitize"
+ case "$mval" in
+ true) mval="address,undefined" ;;
+ false) mval="none" ;;
+ esac
+ ;;
*)
mfeat="$feat"
;;
diff --git a/debian/etc/paivana/paivana.conf b/debian/etc/paivana/paivana.conf
@@ -17,6 +17,32 @@ SERVE = systemd
# if left empty. Only used if "SERVE" is 'tcp'.
# BIND_TO =
+# Largest request body accepted; anything above it is answered 413.
+# A policy decision, not a memory one -- bodies are streamed, so this
+# does not bound what paivana holds.
+# MAX_REQUEST_SIZE = 1048576
+
+# Bytes of a body held in memory at once while it is relayed, per
+# direction per request. Throughput knobs: larger means fewer
+# suspend/resume round trips on a fast link and more memory per request
+# in flight, the worst case being these times CONNECTION_LIMIT. There
+# is deliberately no ceiling on the size of a *response*; bound that at
+# the origin if you want one.
+# REQUEST_BUFFER_MAX = 262144
+# RESPONSE_BUFFER_MAX = 262144
+
+# How long the origin has to produce response headers before we answer
+# 504. Not a bound on the whole request: a large download legitimately
+# runs for longer than any useful ceiling, and after the headers have
+# gone out there is no status left to send anyway.
+# UPSTREAM_TIMEOUT = 60 s
+
+# How long the origin may move no bytes at all, in either direction,
+# before we give up. The clock does not run while paivana is itself
+# holding the origin back because the client has not caught up, so a
+# client on a slow link is not mistaken for a slow origin.
+# UPSTREAM_STALL_TIMEOUT = 60 s
+
# MERCHANT_ACCESS_TOKEN and SECRET live in the file below rather than
# here, because this file is world-readable and those two are not
# things every local account should be able to read. Keep it that way.
diff --git a/meson.build b/meson.build
@@ -145,6 +145,18 @@ if not get_option('only-doc')
dependencies: [gnunetcurl_dep],
required: true,
)
+ # The streaming download API, without which a proxied response
+ # would have to be assembled in memory before any of it could be
+ # forwarded -- and could not exceed GNUNET_MAX_MALLOC_CHECKED at
+ # all. New in GNUnet 0.29.0.
+ if not cc.has_function(
+ 'GNUNET_CURL_job_set_paused',
+ prefix: '#include <gnunet/gnunet_curl_lib.h>',
+ dependencies: [gnunetcurl_dep],
+ )
+ error('libgnunetcurl is too old: GNUNET_CURL_job_add_stream() and '
+ + 'GNUNET_CURL_job_set_paused() are required (GNUnet >= 0.29.0)')
+ endif
talerutil_dep = dependency('talerutil', required: false)
if not talerutil_dep.found()
diff --git a/src/backend/paivana-httpd.c b/src/backend/paivana-httpd.c
@@ -65,7 +65,23 @@ struct GNUNET_STRINGS_IPv6NetworkPolicy *PH_trusted_proxies6;
bool PH_have_trusted_proxies;
-unsigned long long PH_request_buffer_max = 1024 * 1024;
+/**
+ * Compiled-in default for #PH_request_buffer_max. Named because
+ * `run()` has to be able to tell it apart from a value the operator
+ * wrote, which is what decides whether #PH_max_request_size inherits
+ * it.
+ */
+#define PH_DEFAULT_REQUEST_BUFFER_MAX (256 * 1024)
+
+unsigned long long PH_request_buffer_max = PH_DEFAULT_REQUEST_BUFFER_MAX;
+
+unsigned long long PH_response_buffer_max = 256 * 1024;
+
+unsigned long long PH_max_request_size = 1024 * 1024;
+
+struct GNUNET_TIME_Relative PH_upstream_timeout;
+
+struct GNUNET_TIME_Relative PH_upstream_stall_timeout;
unsigned int PH_connection_limit = 512;
@@ -338,29 +354,41 @@ run (void *cls,
const struct GNUNET_CONFIGURATION_Handle *c)
{
char *secret;
+ bool buffer_max_explicit;
(void) cls;
(void) args;
(void) cfgfile;
PH_cfg = c;
+ PH_upstream_timeout
+ = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
+ 60);
+ PH_upstream_stall_timeout
+ = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
+ 60);
GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
NULL);
if ( (0 == PH_request_buffer_max) ||
(PH_request_buffer_max > GNUNET_MAX_MALLOC_CHECKED) )
{
- /* 0 would reject every request that carries a body at all, which
- is not what anyone writing it means; above the hard buffer cap
- the number simply cannot take effect. A negative argument ends
- up here too: GNUnet parses the option with sscanf("%llu"), which
- reads "-1" as ULLONG_MAX rather than complaining. */
+ /* 0 would leave no room to move a body through at all; above the
+ hard allocation cap the buffer simply cannot be made. A negative
+ argument ends up here too: GNUnet parses the option with
+ sscanf("%llu"), which reads "-1" as ULLONG_MAX rather than
+ complaining. */
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"-u/--max-upload must be between 1 and %llu:"
- " request bodies are buffered whole before forwarding\n",
+ " it sizes the buffer a request body is relayed through\n",
(unsigned long long) GNUNET_MAX_MALLOC_CHECKED);
PH_global_ret = EXIT_INVALIDARGUMENT;
GNUNET_SCHEDULER_shutdown ();
return;
}
+ /* `-u' is applied by GNUNET_PROGRAM_run() before we are called, so a
+ value differing from the compiled-in default is one the operator
+ wrote. See PH_max_request_size for why that has to be visible. */
+ buffer_max_explicit = (PH_DEFAULT_REQUEST_BUFFER_MAX
+ != PH_request_buffer_max);
if (! PH_no_check)
{
if (GNUNET_OK !=
@@ -451,6 +479,107 @@ run (void *cls,
}
PH_per_ip_connection_limit = (unsigned int) v;
}
+ if (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_number (c,
+ "paivana",
+ "REQUEST_BUFFER_MAX",
+ &v))
+ {
+ if ( (0 == v) ||
+ (v > GNUNET_MAX_MALLOC_CHECKED) )
+ {
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "paivana",
+ "REQUEST_BUFFER_MAX",
+ "must be between 1 and 40 MiB");
+ PH_global_ret = EXIT_NOTCONFIGURED;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ PH_request_buffer_max = v;
+ buffer_max_explicit = true;
+ }
+ if (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_number (c,
+ "paivana",
+ "RESPONSE_BUFFER_MAX",
+ &v))
+ {
+ if ( (0 == v) ||
+ (v > GNUNET_MAX_MALLOC_CHECKED) )
+ {
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "paivana",
+ "RESPONSE_BUFFER_MAX",
+ "must be between 1 and 40 MiB");
+ PH_global_ret = EXIT_NOTCONFIGURED;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ PH_response_buffer_max = v;
+ }
+ if (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_number (c,
+ "paivana",
+ "MAX_REQUEST_SIZE",
+ &v))
+ {
+ PH_max_request_size = v;
+ }
+ else if (buffer_max_explicit)
+ {
+ /* Before streaming these were one number: the buffer a body was
+ assembled in *was* the largest body we would accept. An
+ operator who raised it to permit large uploads meant the limit,
+ so honour that reading rather than silently tightening their
+ configuration back down to the 1 MiB default. */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Taking MAX_REQUEST_SIZE from the configured request"
+ " buffer size (%llu bytes); set it explicitly to say"
+ " otherwise\n",
+ PH_request_buffer_max);
+ PH_max_request_size = PH_request_buffer_max;
+ }
+ }
+ {
+ struct GNUNET_TIME_Relative st;
+
+ if (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_time (c,
+ "paivana",
+ "UPSTREAM_TIMEOUT",
+ &st))
+ {
+ if (0 == st.rel_value_us)
+ {
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "paivana",
+ "UPSTREAM_TIMEOUT",
+ "must not be zero");
+ PH_global_ret = EXIT_NOTCONFIGURED;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ PH_upstream_timeout = st;
+ }
+ if (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_time (c,
+ "paivana",
+ "UPSTREAM_STALL_TIMEOUT",
+ &st))
+ {
+ if (0 == st.rel_value_us)
+ {
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "paivana",
+ "UPSTREAM_STALL_TIMEOUT",
+ "must not be zero");
+ PH_global_ret = EXIT_NOTCONFIGURED;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ PH_upstream_stall_timeout = st;
+ }
}
{
unsigned int n4;
@@ -863,7 +992,7 @@ main (int argc,
"max-upload",
"BYTES",
gettext_noop (
- "maximum request body size to buffer before forwarding (default: 1048576)"),
+ "bytes of a request body to hold in memory while relaying it upstream (default: 262144); the largest body accepted is MAX_REQUEST_SIZE"),
&PH_request_buffer_max),
GNUNET_GETOPT_OPTION_END
};
diff --git a/src/backend/paivana-httpd.h b/src/backend/paivana-httpd.h
@@ -209,11 +209,16 @@ extern unsigned int PH_connection_limit;
extern unsigned int PH_per_ip_connection_limit;
/**
- * Maximum size (in bytes) of a request body that we will buffer before
- * forwarding it upstream. Requests exceeding this are rejected with
- * HTTP 413. Settable via the `-u` / `--max-upload` command-line
- * option; defaults to 1 MiB and cannot be set above the 40 MiB
- * #GNUNET_MAX_MALLOC_CHECKED the buffering path enforces anyway.
+ * How many bytes of a request body we hold in memory at once while
+ * relaying it upstream, from `REQUEST_BUFFER_MAX` or the `-u` /
+ * `--max-upload` command-line option; 256 KiB by default.
+ *
+ * A throughput knob, not a limit: the body is streamed, so this bounds
+ * only how far the client may run ahead of the origin before we stop
+ * reading from it. What an upload is *allowed* to be is
+ * #PH_max_request_size. Larger means fewer suspend/resume round trips
+ * on a fast link and more memory per request in flight; the worst case
+ * is the product with #PH_connection_limit.
*
* Bounds the *proxied* path only. Bodies sent to our own endpoints
* never reach this buffer: `POST /.well-known/paivana` is read by
@@ -222,5 +227,56 @@ extern unsigned int PH_per_ip_connection_limit;
*/
extern unsigned long long PH_request_buffer_max;
+/**
+ * How many bytes of a response body we hold in memory at once while
+ * relaying it to the client, from `RESPONSE_BUFFER_MAX`; 256 KiB by
+ * default. The mirror of #PH_request_buffer_max, and equally not a
+ * limit — there is deliberately no ceiling on the size of a response.
+ */
+extern unsigned long long PH_response_buffer_max;
+
+/**
+ * Largest request body we accept, from `MAX_REQUEST_SIZE`; 1 MiB by
+ * default. Anything above it is answered 413, on the declared
+ * `Content-Length` where there is one and otherwise once the body
+ * actually exceeds it.
+ *
+ * Distinct from #PH_request_buffer_max, which used to do both jobs
+ * because a body that could not be buffered could not be forwarded.
+ * Streaming separates them: this one is policy, that one is memory.
+ * For the sake of configurations written when they were one number,
+ * an explicitly configured `REQUEST_BUFFER_MAX` with no
+ * `MAX_REQUEST_SIZE` beside it still sets both.
+ */
+extern unsigned long long PH_max_request_size;
+
+/**
+ * How long the origin has to produce its response *headers* before we
+ * give up on it and answer 504, from `UPSTREAM_TIMEOUT`; 60 s by
+ * default.
+ *
+ * Deliberately not a bound on the whole request: a response is
+ * relayed as it arrives, so a large one legitimately runs for as long
+ * as it runs. This clock is cancelled the moment the final header
+ * section ends, and it is the only one of the three that can still
+ * produce a status code — after that the status is already on the
+ * wire. A stalling origin is caught afterwards by
+ * #PH_upstream_stall_timeout instead.
+ */
+extern struct GNUNET_TIME_Relative PH_upstream_timeout;
+
+/**
+ * How long the origin may make no progress at all — no byte moved in
+ * either direction — before we give up on the request, from
+ * `UPSTREAM_STALL_TIMEOUT`; 60 s by default.
+ *
+ * Not a bound on how long a request may take: a 500 MiB download is
+ * expected to run for as long as it runs. The clock is suspended
+ * whenever *we* are the reason nothing is moving, i.e. while we hold
+ * libcurl's receive side paused because the client has not drained
+ * what we already have.
+ */
+extern struct GNUNET_TIME_Relative PH_upstream_stall_timeout;
+
#endif
diff --git a/src/backend/paivana-httpd_reverse.c b/src/backend/paivana-httpd_reverse.c
@@ -88,115 +88,151 @@
"GET, HEAD, POST, PUT, PATCH, DELETE, OPTIONS"
/**
- * How long the origin has to complete a response before we give up on
- * it and answer 504.
+ * Value for CURLOPT_CONNECTTIMEOUT, in seconds.
*
- * This is the *effective* per-request ceiling, and the only one there
- * is. MHD_OPTION_CONNECTION_TIMEOUT (see `paivana-httpd_daemon.c')
- * does not bound the phase a proxied request actually spends waiting:
- * MHD removes a suspended connection from its timeout lists, and every
- * request is suspended for exactly as long as the upstream takes. So
- * changing the MHD timeout has no effect on how long a slow origin can
- * hold a connection open; this constant, and #PH_CURL_TIMEOUT_S behind
- * it, are what do.
+ * Must stay below #PH_upstream_timeout: a connect that hangs is a 502,
+ * and it has to be able to say so before the gateway-timeout deadline
+ * claims it and calls the same failure a 504.
*/
-#define PH_UPSTREAM_TIMEOUT \
- GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 60)
+#define PH_CURL_CONNECT_TIMEOUT_S 30L
+
/**
- * Value for CURLOPT_TIMEOUT, in seconds.
- *
- * Deliberately *longer* than #PH_UPSTREAM_TIMEOUT rather than equal to
- * it. libcurl reports a timeout the same way it reports a refused
- * connection or a TLS failure -- CURLINFO_RESPONSE_CODE of 0, with the
- * CURLcode not surfaced by libgnunetcurl -- so the only way to tell
- * "the origin was slow" (504, and worth retrying) from "the origin was
- * not there" (502, and not) is to have our own deadline expire first.
- * Setting both to 60 s would make which of them fires a race. This one
- * remains as a backstop: if it ever fires, our task did not run, and
- * 502 is then the honest answer.
+ * Size of the buffer MHD fills from our content reader, and the unit
+ * in which a chunked response is chunked. Allocated per queued
+ * response, so it is a per-in-flight-request cost on top of
+ * #PH_response_buffer_max and wants to stay modest; 16 KiB also
+ * matches CURL_MAX_WRITE_SIZE, the most libcurl hands us at once, so
+ * one MHD read tends to correspond to one libcurl write.
*/
-#define PH_CURL_TIMEOUT_S 90L
+#define PH_STREAM_BLOCK_SIZE (16 * 1024)
/**
- * Value for CURLOPT_CONNECTTIMEOUT, in seconds.
- *
- * Must stay below #PH_UPSTREAM_TIMEOUT for the same reason: a connect
- * that hangs is a 502, and it has to be able to say so before the
- * gateway-timeout deadline claims it.
+ * Smallest ring we will allocate, and the step the ring grows in.
+ * Also the floor under the configured caps: a ring that cannot hold
+ * one libcurl delivery would pause with an empty buffer and never be
+ * woken, so #ring_put() has to be able to take CURL_MAX_WRITE_SIZE.
*/
-#define PH_CURL_CONNECT_TIMEOUT_S 30L
+#define PH_RING_MIN_SIZE (16 * 1024)
/**
- * State machine for HTTP requests (per request). MHD invokes the
- * access handler multiple times per request — we use this enum to
- * know what each invocation is expected to do, and cascade between
- * states in a single invocation when no new data from MHD is
- * required.
+ * How far the client's request body has got.
+ *
+ * Independent of #DownState: since the origin request now starts
+ * before the body has arrived, the two directions run at the same
+ * time and neither is a stage of the other.
*/
-enum RequestState
+enum UploadState
{
/**
* Initial state. MHD's first access-handler call (immediately
* after parsing the request headers, `upload_data_size == 0`)
- * has not yet been observed. In this state we can still queue
- * a final response before MHD auto-generates a 100 Continue and
- * the client invariably starts with the upload (if any).
+ * has not yet been observed. In this state we can still queue a
+ * final response before MHD auto-generates a 100 Continue and the
+ * client invariably starts with the upload (if any).
*/
- REQUEST_STATE_HEADERS_PENDING,
+ UP_HEADERS_PENDING,
/**
- * We have accepted the request and are receiving the
- * body from the client. Each body chunk is buffered into
- * `io_buf`; when MHD signals end-of-body (size == 0) we
- * advance to `CLIENT_UPLOAD_DONE`.
+ * Relaying the client's body: MHD hands us chunks, they go into
+ * @e up_ring, libcurl's read callback takes them out.
*/
- REQUEST_STATE_CLIENT_UPLOAD_STARTED,
+ UP_STREAMING,
/**
- * We have decided to reject the upload (Content-Length exceeded
- * the buffer cap, or the actual body did) but MHD is still
- * delivering body chunks and refuses to let us queue a response
- * during `BODY_RECEIVING`. Silently drop the bytes here until
- * MHD gets back to a state that allows `MHD_queue_response`.
+ * The client's body is fully received. Whatever is still in
+ * @e up_ring has to reach the origin, after which the read
+ * callback reports end-of-body.
*/
- REQUEST_STATE_REJECT_UPLOAD_DRAIN,
+ UP_CLIENT_DONE,
/**
- * Ready to queue the 413 Content Too Large response.
+ * The origin has consumed the whole body.
*/
- REQUEST_STATE_REJECT_UPLOAD,
+ UP_DONE,
/**
- * Client body fully received; next step is to initialize the
- * curl handle and start forwarding to the upstream.
+ * The origin transfer is over — it answered early, or failed —
+ * while the client is still sending. Its remaining bytes are read
+ * and discarded, because MHD will not let us queue a response
+ * while the request body is still arriving and closing the
+ * connection instead would throw away the status the origin took
+ * the trouble to send.
*/
- REQUEST_STATE_CLIENT_UPLOAD_DONE,
+ UP_DRAINING
+};
+
+/**
+ * How far the origin's response has got.
+ */
+enum DownState
+{
/**
- * We have started uploading data to the proxied service.
- * MHD handling will be suspended.
+ * The final response's header section has not ended yet. Nothing
+ * can be queued to the client: the status is not known.
*/
- REQUEST_STATE_PROXY_UPLOAD_STARTED,
+ DOWN_WAITING_HEADERS,
/**
- * We've finished uploading data via CURL and can now download.
- * MHD handling should remain suspended.
+ * Headers are in and the response is (or is about to be) queued;
+ * body bytes are moving from libcurl through @e down_ring into
+ * MHD's content reader.
*/
- REQUEST_STATE_PROXY_DOWNLOAD_STARTED,
+ DOWN_STREAMING,
/**
- * We've finished receiving download data from cURL; ready to
- * build and queue the final response to the client.
+ * The origin delivered its whole body. Once @e down_ring drains,
+ * the content reader reports end of stream.
*/
- REQUEST_STATE_PROXY_DOWNLOAD_DONE,
+ DOWN_COMPLETE,
/**
- * We've failed to download data from cURL; ready to
- * build and queue the final response to the client.
+ * The transfer failed. If this happened before #DOWN_STREAMING we
+ * still owe the client a status (502 or 504); if it happened after,
+ * the status is long gone and all we can do is break the framing.
*/
- REQUEST_STATE_PROXY_DOWNLOAD_FAILED
+ DOWN_FAILED
+};
+
+
+/**
+ * One direction's worth of body bytes in flight, as a ring buffer.
+ *
+ * A ring rather than a list of chunks: one allocation instead of one
+ * per delivery, a hard bound with no fragmentation, and no per-chunk
+ * bookkeeping. Grown lazily from #PH_RING_MIN_SIZE so that a 3 KiB
+ * HTML page does not cost a quarter megabyte — which matters now that
+ * every response takes this path, not only the large ones.
+ */
+struct PH_Ring
+{
+ /**
+ * The bytes, @e size of them allocated, or NULL before the first
+ * one arrives.
+ */
+ char *buf;
+
+ /**
+ * Bytes allocated in @e buf.
+ */
+ size_t size;
+
+ /**
+ * Bytes currently held, starting at @e off and wrapping.
+ */
+ size_t len;
+
+ /**
+ * Offset of the oldest byte held.
+ */
+ size_t off;
+
+ /**
+ * Largest @e size we will grow to on demand.
+ */
+ size_t cap;
};
@@ -284,48 +320,79 @@ struct HttpRequest
struct HttpResponseHeader *header_tail;
/**
- * Buffer we use for moving data between MHD and
- * curl (in both directions).
+ * Client's request body on its way to the origin.
*/
- char *io_buf;
+ struct PH_Ring up_ring;
/**
- * Number of bytes still to be sent from the IO buffer (upload
- * phase), or number of bytes already in the IO buffer (while the
- * client body is still being received).
+ * Origin's response body on its way to the client.
*/
- size_t io_len;
+ struct PH_Ring down_ring;
/**
- * Read cursor into `io_buf` for the upload-to-curl phase: the next
- * byte to hand to libcurl is `io_buf[io_off]`.
+ * Bytes of request body received from the client so far, checked
+ * against #PH_max_request_size. A client that declares no
+ * `Content-Length` can only be caught here.
*/
- size_t io_off;
+ uint64_t up_received;
/**
- * Number of bytes allocated for the IO buffer.
+ * Bytes of response body handed to MHD so far. Only used for
+ * logging a truncated response.
*/
- unsigned int io_size;
+ uint64_t down_sent;
/**
- * HTTP response code to give to MHD for the response.
+ * HTTP response code to give to MHD for the response. Taken from
+ * the origin's status line in `curl_check_hdr()` rather than from
+ * CURLINFO_RESPONSE_CODE, since it is needed while the transfer is
+ * still running.
*/
unsigned int response_code;
/**
* `Content-Length` the upstream declared for its response, or
* #PH_NO_CONTENT_LENGTH if it declared none we could parse.
- * `curl_check_hdr()` drops the header itself — the length we
- * forward has to describe the buffer we actually assembled — but
- * the declared value is kept here because it is the only evidence
- * that the buffer is short; see `curl_download_cb()`.
+ * `curl_check_hdr()` drops the header itself and this value becomes
+ * the size of the MHD response instead, which is what makes MHD
+ * emit it again — and what makes a HEAD or a 304 report the length
+ * of a body it does not send.
*/
uint64_t upstream_content_length;
/**
- * Request processing state machine.
+ * How far the client's request body has got.
+ */
+ enum UploadState up_state;
+
+ /**
+ * How far the origin's response has got.
+ */
+ enum DownState down_state;
+
+ /**
+ * Deferred unpause of libcurl, or NULL if none is pending.
+ *
+ * Never done directly from an MHD callback: GNUNET_CURL_job_unpause()
+ * hands over whatever libcurl buffered while paused by calling
+ * straight back into our stream callback, which would then be
+ * running inside MHD's content reader and could resume a connection
+ * that is not suspended. One scheduler round trip of latency
+ * removes the whole class of re-entrancy bug.
+ */
+ struct GNUNET_SCHEDULER_Task *unpause_task;
+
+ /**
+ * Watchdog for #PH_upstream_stall_timeout; see `stall_check()`.
*/
- enum RequestState state;
+ struct GNUNET_SCHEDULER_Task *stall_task;
+
+ /**
+ * When a byte last moved in either direction. The stall watchdog
+ * measures from here rather than being rearmed on every delivery,
+ * which would mean a scheduler cancel-and-add per 16 KiB.
+ */
+ struct GNUNET_TIME_Absolute last_progress;
/**
* Did we suspend MHD processing?
@@ -389,14 +456,61 @@ struct HttpRequest
bool interim_response;
/**
- * Did #PH_UPSTREAM_TIMEOUT expire on this request? Decides whether
- * #REQUEST_STATE_PROXY_DOWNLOAD_FAILED becomes a 504 or a 502.
+ * Did #PH_upstream_timeout or #PH_upstream_stall_timeout expire on
+ * this request? Decides whether a #DOWN_FAILED that still owes the
+ * client a status becomes a 504 or a 502.
*/
bool upstream_timed_out;
/**
+ * Are we holding libcurl's receive side paused because @e down_ring
+ * is full? While this is set the stall watchdog does not run: the
+ * origin is not sending because we told it not to.
+ */
+ bool curl_recv_paused;
+
+ /**
+ * Are we holding libcurl's send side paused because @e up_ring is
+ * empty and the client has not sent the rest of its body?
+ */
+ bool curl_send_paused;
+
+ /**
+ * Directions @e unpause_task is to resume.
+ */
+ bool want_recv_cont;
+
+ /**
+ * Directions @e unpause_task is to resume.
+ */
+ bool want_send_cont;
+
+ /**
+ * Has the client's body been refused (413)? Set on the declared
+ * `Content-Length` where there is one, and otherwise once the body
+ * actually exceeds #PH_max_request_size.
+ */
+ bool reject_upload;
+
+ /**
+ * Has `PAIVANA_HTTPD_reverse_cleanup()` run? Half of the ownership
+ * handshake with @e reader_gone: MHD calls its completion notifier
+ * *before* it destroys the response, so at that moment MHD still
+ * holds a content reader whose closure is this very structure.
+ * Whoever finishes second frees it.
+ */
+ bool mhd_gone;
+
+ /**
+ * Has the content reader's free callback run, or was one never
+ * armed? Starts true, and is cleared when the streaming response
+ * is created. See @e mhd_gone.
+ */
+ bool reader_gone;
+
+ /**
* Deadline for the upstream request, armed while @e job is
- * outstanding. See #PH_UPSTREAM_TIMEOUT.
+ * outstanding. See #PH_upstream_timeout.
*/
struct GNUNET_SCHEDULER_Task *timeout_task;
@@ -1022,28 +1136,465 @@ free_response_headers (struct HttpRequest *hr)
}
+/* ******************* ring buffers ******************** */
+
+
+/**
+ * Release the memory @a r holds.
+ *
+ * @param[in,out] r ring to empty
+ */
+static void
+ring_free (struct PH_Ring *r)
+{
+ GNUNET_free (r->buf);
+ r->size = 0;
+ r->len = 0;
+ r->off = 0;
+}
+
+
+/**
+ * Make room for at least @a need bytes in @a r, if we may.
+ *
+ * Doubles from #PH_RING_MIN_SIZE and stops at @e cap, so a small
+ * response never costs more than the one delivery it arrived in.
+ * Linearises on the way, which is what keeps the wrap arithmetic in
+ * `ring_put()` and `ring_get()` to a single branch each.
+ *
+ * @param[in,out] r ring to grow
+ * @param need bytes the caller wants to be able to hold in total
+ * @param beyond_cap grow past @e cap if that is what @a need takes;
+ * for the one caller that would otherwise deadlock
+ */
+static void
+ring_grow (struct PH_Ring *r,
+ size_t need,
+ bool beyond_cap)
+{
+ size_t ns;
+ char *nb;
+
+ if ( (! beyond_cap) &&
+ (need > r->cap) )
+ need = r->cap;
+ if (need <= r->size)
+ return;
+ ns = (0 == r->size) ? PH_RING_MIN_SIZE : r->size;
+ while (ns < need)
+ ns *= 2;
+ if ( (! beyond_cap) &&
+ (ns > r->cap) )
+ ns = r->cap;
+ /* Deliberately not GNUNET_malloc(): that aborts the whole daemon
+ when malloc() returns NULL, dropping every other client because
+ one body could not be relayed. Keeping the smaller ring instead
+ merely means more suspend/resume round trips. */
+ nb = GNUNET_malloc_large (ns);
+ if (NULL == nb)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Could not grow a relay buffer to %llu bytes\n",
+ (unsigned long long) ns);
+ return;
+ }
+ if (0 != r->len)
+ {
+ size_t first = GNUNET_MIN (r->len,
+ r->size - r->off);
+
+ GNUNET_memcpy (nb,
+ &r->buf[r->off],
+ first);
+ if (first < r->len)
+ GNUNET_memcpy (&nb[first],
+ r->buf,
+ r->len - first);
+ }
+ GNUNET_free (r->buf);
+ r->buf = nb;
+ r->size = ns;
+ r->off = 0;
+}
+
+
+/**
+ * Copy as much of @a data into @a r as fits, growing it if it may.
+ *
+ * @param[in,out] r ring to append to
+ * @param data bytes to append
+ * @param n number of bytes in @a data
+ * @param beyond_cap see `ring_grow()`
+ * @return number of bytes taken, 0 if there was no room at all
+ */
+static size_t
+ring_put (struct PH_Ring *r,
+ const char *data,
+ size_t n,
+ bool beyond_cap)
+{
+ size_t wpos;
+ size_t first;
+
+ if (r->size - r->len < n)
+ ring_grow (r,
+ r->len + n,
+ beyond_cap);
+ if (n > r->size - r->len)
+ n = r->size - r->len;
+ if (0 == n)
+ return 0;
+ wpos = (r->off + r->len) % r->size;
+ first = GNUNET_MIN (n,
+ r->size - wpos);
+ GNUNET_memcpy (&r->buf[wpos],
+ data,
+ first);
+ if (first < n)
+ GNUNET_memcpy (r->buf,
+ &data[first],
+ n - first);
+ r->len += n;
+ return n;
+}
+
+
+/**
+ * Copy all of @a data into @a r, or nothing at all.
+ *
+ * The all-or-nothing part is load-bearing on the download side.
+ * libcurl's write callback cannot take a delivery partially — any
+ * short return fails the transfer — so the only other answer is to
+ * pause, and after the resume libcurl offers *the same bytes again
+ * from the start*. A partial copy followed by a pause therefore
+ * writes its prefix into the ring twice, which shows up as a
+ * response that is longer than the origin sent and corrupt from the
+ * first pause onwards.
+ *
+ * @param[in,out] r ring to append to
+ * @param data bytes to append
+ * @param n number of bytes in @a data
+ * @param beyond_cap see `ring_grow()'
+ * @return true if all @a n bytes were taken
+ */
+static bool
+ring_put_all (struct PH_Ring *r,
+ const char *data,
+ size_t n,
+ bool beyond_cap)
+{
+ if (r->size - r->len < n)
+ ring_grow (r,
+ r->len + n,
+ beyond_cap);
+ if (r->size - r->len < n)
+ return false; /* nothing copied */
+ GNUNET_assert (n == ring_put (r,
+ data,
+ n,
+ beyond_cap));
+ return true;
+}
+
+
+/**
+ * Copy up to @a n bytes out of @a r into @a out.
+ *
+ * @param[in,out] r ring to consume from
+ * @param[out] out where to put them
+ * @param n capacity of @a out
+ * @return number of bytes copied, 0 if @a r was empty
+ */
+static size_t
+ring_get (struct PH_Ring *r,
+ char *out,
+ size_t n)
+{
+ size_t first;
+
+ if (n > r->len)
+ n = r->len;
+ if (0 == n)
+ return 0;
+ first = GNUNET_MIN (n,
+ r->size - r->off);
+ GNUNET_memcpy (out,
+ &r->buf[r->off],
+ first);
+ if (first < n)
+ GNUNET_memcpy (&out[first],
+ r->buf,
+ n - first);
+ r->off = (r->off + n) % r->size;
+ r->len -= n;
+ if (0 == r->len)
+ r->off = 0; /* keep the next fill linear */
+ return n;
+}
+
+
+/* ************ waking the two halves up ************* */
+
+
+/**
+ * Let MHD get on with @a hr: resume the connection if we suspended
+ * it, and make sure the daemon is polled again.
+ *
+ * Safe to call from a libcurl callback — MHD_resume_connection() is
+ * documented as callable from any thread and TALER_MHD_daemon_trigger()
+ * only re-arms a scheduler task. The reverse direction, unpausing
+ * libcurl, is *not* safe that way; see @e unpause_task.
+ *
+ * @param[in,out] hr request whose client connection to wake
+ */
+static void
+wake_client (struct HttpRequest *hr)
+{
+ if (GNUNET_YES != hr->suspended)
+ return;
+ hr->suspended = GNUNET_NO;
+ MHD_resume_connection (hr->con);
+ TALER_MHD_daemon_trigger ();
+}
+
+
+/**
+ * Stop letting MHD call us about @a hr until something wakes it.
+ *
+ * @param[in,out] hr request whose client connection to park
+ */
+static void
+suspend_client (struct HttpRequest *hr)
+{
+ if (GNUNET_YES == hr->suspended)
+ return;
+ MHD_suspend_connection (hr->con);
+ hr->suspended = GNUNET_YES;
+}
+
+
+/**
+ * Resume the libcurl directions that were asked for.
+ *
+ * @param cls the `struct HttpRequest *`
+ */
+static void
+do_unpause (void *cls)
+{
+ struct HttpRequest *hr = cls;
+ bool woke = false;
+
+ hr->unpause_task = NULL;
+ if (hr->want_recv_cont)
+ {
+ hr->want_recv_cont = false;
+ hr->curl_recv_paused = false;
+ woke = true;
+ }
+ if (hr->want_send_cont)
+ {
+ hr->want_send_cont = false;
+ hr->curl_send_paused = false;
+ woke = true;
+ }
+ if ( (! woke) ||
+ (NULL == hr->job) )
+ return;
+ /* The origin was never at fault for the interval we held it paused,
+ so the stall clock restarts from the resume rather than from the
+ last byte that moved. */
+ hr->last_progress = GNUNET_TIME_absolute_get ();
+ /* May call straight back into our stream or read callback with what
+ libcurl buffered while paused, which is precisely why this is a
+ task and not a call from inside MHD -- and why the two flags are
+ cleared before the call rather than after. */
+ GNUNET_CURL_job_set_paused (hr->job,
+ hr->curl_recv_paused,
+ hr->curl_send_paused);
+}
+
+
+/**
+ * Arrange for libcurl's receive side to be resumed.
+ *
+ * @param[in,out] hr request to resume
+ */
+static void
+schedule_recv_cont (struct HttpRequest *hr)
+{
+ if (! hr->curl_recv_paused)
+ return;
+ hr->want_recv_cont = true;
+ if (NULL == hr->unpause_task)
+ hr->unpause_task = GNUNET_SCHEDULER_add_now (&do_unpause,
+ hr);
+}
+
+
+/**
+ * Arrange for libcurl's send side to be resumed.
+ *
+ * @param[in,out] hr request to resume
+ */
+static void
+schedule_send_cont (struct HttpRequest *hr)
+{
+ if (! hr->curl_send_paused)
+ return;
+ hr->want_send_cont = true;
+ if (NULL == hr->unpause_task)
+ hr->unpause_task = GNUNET_SCHEDULER_add_now (&do_unpause,
+ hr);
+}
+
+
+/* **************** the stall watchdog **************** */
+
+
+/**
+ * Give up on @a hr because the origin transfer failed or stalled.
+ *
+ * @param[in,out] hr request to fail
+ */
+static void
+fail_transfer (struct HttpRequest *hr)
+{
+ if (NULL != hr->job)
+ {
+ GNUNET_CURL_job_cancel (hr->job);
+ hr->job = NULL;
+ }
+ hr->down_state = DOWN_FAILED;
+ if (UP_DONE != hr->up_state)
+ hr->up_state = UP_DRAINING;
+ wake_client (hr);
+}
+
+
+static void
+stall_check (void *cls);
+
+
+/**
+ * Arm the stall watchdog for @a hr, unless we are the reason nothing
+ * is moving.
+ *
+ * @param[in,out] hr request to watch
+ */
+static void
+stall_arm (struct HttpRequest *hr)
+{
+ if (NULL != hr->stall_task)
+ return;
+ if (NULL == hr->job)
+ return;
+ hr->stall_task = GNUNET_SCHEDULER_add_delayed (PH_upstream_stall_timeout,
+ &stall_check,
+ hr);
+}
+
+
+/**
+ * Note that a byte moved on @a hr, in either direction.
+ *
+ * Records the time instead of rearming the watchdog, so that a 200 MiB
+ * transfer does not cost a scheduler cancel-and-add per 16 KiB.
+ *
+ * @param[in,out] hr request that made progress
+ */
+static void
+note_progress (struct HttpRequest *hr)
+{
+ hr->last_progress = GNUNET_TIME_absolute_get ();
+ stall_arm (hr);
+}
+
+
+/**
+ * Task checking whether the origin has made any progress within
+ * #PH_upstream_stall_timeout.
+ *
+ * The clock does not run while we hold libcurl's receive side paused:
+ * the origin has stopped sending because we told it to, and blaming it
+ * for that would kill exactly the transfers this whole change exists
+ * to support — a large download to a client on a slow link. The send
+ * side is different: there, nothing moving means the *client* has gone
+ * quiet mid-body, and since a suspended connection is off MHD's
+ * timeout lists (see `CLIENT_CONNECTION_TIMEOUT`) this watchdog is the
+ * only thing that will ever notice.
+ *
+ * libcurl's own CURLOPT_LOW_SPEED_LIMIT / _TIME would be this, already
+ * written and already tested — and it is deliberately not used.
+ * `Curl_speedcheck()` exempts a transfer whose *receive* side is
+ * paused ("A paused transfer is not qualified for speed checks"), but
+ * there is no matching exemption for the send side, while the speed it
+ * checks is the sum over both directions of one easy handle. Measured
+ * against libcurl 8.14.1: a receive-paused transfer survives a 15 s
+ * pause and a 200 B/s client for 180 s, but a send-paused upload dies
+ * at exactly LOW_SPEED_TIME. Since one handle carries both directions
+ * and the option is one setting on it, enabling it at all would make a
+ * client on a slow uplink kill its own upload.
+ *
+ * @param cls the `struct HttpRequest *`
+ */
+static void
+stall_check (void *cls)
+{
+ struct HttpRequest *hr = cls;
+ struct GNUNET_TIME_Relative rem;
+
+ hr->stall_task = NULL;
+ if (NULL == hr->job)
+ return;
+ if (hr->curl_recv_paused)
+ {
+ /* Our doing, not the origin's. Look again later. */
+ hr->last_progress = GNUNET_TIME_absolute_get ();
+ hr->stall_task = GNUNET_SCHEDULER_add_delayed (PH_upstream_stall_timeout,
+ &stall_check,
+ hr);
+ return;
+ }
+ rem = GNUNET_TIME_absolute_get_remaining (
+ GNUNET_TIME_absolute_add (hr->last_progress,
+ PH_upstream_stall_timeout));
+ if (0 != rem.rel_value_us)
+ {
+ hr->stall_task = GNUNET_SCHEDULER_add_delayed (rem,
+ &stall_check,
+ hr);
+ return;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Upstream `%s' moved no data for %s; giving up\n",
+ hr->url,
+ GNUNET_STRINGS_relative_time_to_string (
+ PH_upstream_stall_timeout,
+ true));
+ hr->upstream_timed_out = true;
+ fail_transfer (hr);
+}
+
+
/**
* Transform _one_ CURL header (gotten from the request) into
* MHD format and put it into the response headers list; mostly
* copies the headers, but makes special adjustments based on
* control requests.
*
+ * @param cls our `struct HttpRequest *`
* @param buffer curl buffer with a single
* line of header data; not 0-terminated!
- * @param size curl blocksize
- * @param nmemb curl blocknumber
- * @param cls our `struct HttpRequest *`
- * @return size of processed bytes
+ * @param bytes number of bytes in @a buffer
+ * @return #GNUNET_OK to carry on with the transfer
*/
-static size_t
-curl_check_hdr (void *buffer,
- size_t size,
- size_t nmemb,
- void *cls)
+static enum GNUNET_GenericReturnValue
+curl_check_hdr (void *cls,
+ const char *buffer,
+ size_t bytes)
{
struct HttpRequest *hr = cls;
struct HttpResponseHeader *header;
- size_t bytes = size * nmemb;
size_t len;
char *ndup;
const char *hdr_type;
@@ -1053,7 +1604,7 @@ curl_check_hdr (void *buffer,
/* Past the end of the final header section: a trailer field. See
@e headers_complete. */
if (hr->headers_complete)
- return bytes;
+ return GNUNET_OK;
/* Raw line is not guaranteed to be null-terminated. */
ndup = GNUNET_malloc (bytes + 1);
memcpy (ndup,
@@ -1074,11 +1625,32 @@ curl_check_hdr (void *buffer,
decides whether trailers follow (final) or another response is
still to come (interim). */
if (hr->interim_response)
+ {
hr->interim_response = false;
- else
- hr->headers_complete = true;
+ GNUNET_free (ndup);
+ return GNUNET_OK;
+ }
+ hr->headers_complete = true;
GNUNET_free (ndup);
- return bytes;
+ /* The status is known, the header list is complete, and any
+ `Content-Length' the origin declared has been recorded: this is
+ the earliest moment at which the client can be given a status
+ line. Everything after it is body, relayed as it arrives.
+
+ The response is not queued here. We are inside a libcurl
+ callback, and queuing is MHD's business — and in any case MHD
+ will not accept a response while the client's own request body
+ is still coming in. Waking the access handler lets it decide;
+ see `queue_stream_response()'. */
+ hr->down_state = DOWN_STREAMING;
+ if (NULL != hr->timeout_task)
+ {
+ GNUNET_SCHEDULER_cancel (hr->timeout_task);
+ hr->timeout_task = NULL;
+ }
+ note_progress (hr);
+ wake_client (hr);
+ return GNUNET_OK;
}
if (0 == strncasecmp (ndup,
"HTTP/",
@@ -1108,22 +1680,26 @@ curl_check_hdr (void *buffer,
hr->upstream_content_length = PH_NO_CONTENT_LENGTH;
GNUNET_free (hr->upstream_connection);
hr->upstream_connection = NULL;
+ /* Taken from the status line rather than from
+ CURLINFO_RESPONSE_CODE, which libgnunetcurl only reads out at
+ completion: the response is queued long before that. */
+ hr->response_code = code;
}
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
if (hr->interim_response)
{
PAIVANA_LOG_DEBUG ("Not merging interim-response header line `%s'\n",
ndup);
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
hdr_type = strtok (ndup, ":");
if (NULL == hdr_type)
{
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
if (! is_field_name (hdr_type))
{
@@ -1137,13 +1713,13 @@ curl_check_hdr (void *buffer,
hr->url,
hdr_type);
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
hdr_val = strtok (NULL, "\n");
if (NULL == hdr_val)
{
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
while (' ' == *hdr_val)
hdr_val++;
@@ -1159,11 +1735,12 @@ curl_check_hdr (void *buffer,
PAIVANA_LOG_DEBUG ("Parsed line: '%s: %s'\n",
hdr_type,
hdr_val);
- /* Skip "Content-length:" header as it will be wrong, given
- that we are man-in-the-middling the connection. Remember the
- value first: it is what lets `curl_download_cb()' tell a complete
- body from one libgnunetcurl stopped buffering (RFC 9110 §8.6 —
- the field states the length of the body that follows). */
+ /* Do not relay the "Content-Length:" line itself: MHD emits the
+ field from the size the response was created with, and adding a
+ second one is a framing error. The value becomes exactly that
+ size in `queue_stream_response()', so the origin's number does
+ reach the client — including on a HEAD or a 304, where it states
+ the length of a body that is not sent (RFC 9110 §8.6). */
if (0 == strcasecmp (hdr_type,
MHD_HTTP_HEADER_CONTENT_LENGTH))
{
@@ -1179,7 +1756,7 @@ curl_check_hdr (void *buffer,
('\0' == *endptr) )
hr->upstream_content_length = (uint64_t) cl;
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
/* Remember the upstream's Connection list; the headers it names
are connection-specific and get stripped from `header_head' in
@@ -1195,7 +1772,7 @@ curl_check_hdr (void *buffer,
if (is_hop_by_hop_header (hdr_type))
{
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
if (0 != strlen (hdr_val)) /* Rely in MHD to set those */
{
@@ -1211,7 +1788,7 @@ curl_check_hdr (void *buffer,
header);
}
GNUNET_free (ndup);
- return bytes;
+ return GNUNET_OK;
}
@@ -1236,15 +1813,19 @@ status_allows_body (unsigned int code)
/**
- * Task run when the origin did not complete a response within
- * #PH_UPSTREAM_TIMEOUT.
+ * Task run when the origin did not produce response *headers* within
+ * #PH_upstream_timeout.
*
* Exists so that a slow origin can be told apart from an unreachable
- * one: both reach `curl_download_cb()' as a response code of 0, and
- * RFC 9110 gives them different status codes (§15.6.5 504 "did not
- * receive a timely response from an upstream server" versus §15.6.3
- * 502). Monitoring and CDN retry policies act on the difference -- a
- * 504 is retried, a 502 usually is not.
+ * one: both reach `curl_done_cb()' as a response code of 0, and RFC
+ * 9110 gives them different status codes (§15.6.5 504 "did not receive
+ * a timely response from an upstream server" versus §15.6.3 502).
+ * Monitoring and CDN retry policies act on the difference -- a 504 is
+ * retried, a 502 usually is not.
+ *
+ * Cancelled once the header section ends, because from there on the
+ * status is already on its way to the client and a deadline can no
+ * longer produce one. `stall_check()' takes over.
*
* @param cls the `struct HttpRequest *` that ran out of time
*/
@@ -1258,231 +1839,181 @@ upstream_timeout (void *cls)
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
"Upstream did not answer for `%s' within %s\n",
hr->url,
- GNUNET_STRINGS_relative_time_to_string (PH_UPSTREAM_TIMEOUT,
+ GNUNET_STRINGS_relative_time_to_string (PH_upstream_timeout,
true));
- if (NULL != hr->job)
- {
- GNUNET_CURL_job_cancel (hr->job);
- hr->job = NULL;
- }
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_FAILED;
- if (GNUNET_YES == hr->suspended)
- {
- hr->suspended = GNUNET_NO;
- MHD_resume_connection (hr->con);
- TALER_MHD_daemon_trigger ();
+ fail_transfer (hr);
+}
+
+
+/**
+ * Handle a chunk of the response body as it arrives from the origin.
+ *
+ * libcurl's write callback semantics leave no room for a partial take:
+ * anything short of @a data_size fails the transfer. So the ring
+ * either has room for the whole delivery or we pause and libcurl
+ * offers the same bytes again after the resume.
+ *
+ * @param cls our `struct HttpRequest *`
+ * @param data body bytes
+ * @param data_size number of bytes in @a data
+ * @return @a data_size, or #GNUNET_CURL_STREAM_PAUSE
+ */
+static size_t
+stream_body_cb (void *cls,
+ const void *data,
+ size_t data_size)
+{
+ struct HttpRequest *hr = cls;
+
+ if (0 == data_size)
+ return 0;
+ if (DOWN_FAILED == hr->down_state)
+ return 0; /* fail the transfer; we are already giving up */
+ if (! ring_put_all (&hr->down_ring,
+ data,
+ data_size,
+ false))
+ {
+ if (0 != hr->down_ring.len)
+ {
+ /* The client has not drained what we already hold. Stop reading
+ from the origin until it does -- this is the backpressure the
+ whole design turns on, and the reason the stall watchdog does
+ not run while it is in effect. Nothing has been taken from
+ @a data; libcurl offers it again after the resume. */
+ hr->curl_recv_paused = true;
+ return GNUNET_CURL_STREAM_PAUSE;
+ }
+ /* Empty and still too small: the configured cap is below one
+ libcurl delivery, or the allocation failed. Pausing here would
+ deadlock, since nothing will ever free more room. */
+ if (! ring_put_all (&hr->down_ring,
+ data,
+ data_size,
+ true))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Could not buffer %llu bytes of the response to `%s'\n",
+ (unsigned long long) data_size,
+ hr->url);
+ return 0; /* fail the transfer */
+ }
}
+ note_progress (hr);
+ wake_client (hr);
+ return data_size;
}
/**
- * Handle response payload data from cURL.
+ * The origin transfer is over, one way or the other.
*
* @param cls our `struct HttpRequest *`
- * @param response_code HTTP status returned by the server
- * @param body response body
- * @param body_size number of blocks of data
+ * @param response_code HTTP status the origin returned, 0 if the
+ * transfer never got that far
+ * @param body unused (NULL for a streaming job)
+ * @param body_size unused (0 for a streaming job)
*/
static void
-curl_download_cb (void *cls,
- long response_code,
- const void *body,
- size_t body_size)
+curl_done_cb (void *cls,
+ long response_code,
+ const void *body,
+ size_t body_size)
{
struct HttpRequest *hr = cls;
+ (void) body;
+ (void) body_size;
+ /* libgnunetcurl frees the easy handle immediately after we return,
+ so anything that might unpause it has to stop referring to it
+ now. */
hr->job = NULL;
if (NULL != hr->timeout_task)
{
GNUNET_SCHEDULER_cancel (hr->timeout_task);
hr->timeout_task = NULL;
}
- if (GNUNET_YES == hr->suspended)
+ if (NULL != hr->stall_task)
{
- hr->suspended = GNUNET_NO;
- MHD_resume_connection (hr->con);
- TALER_MHD_daemon_trigger ();
+ GNUNET_SCHEDULER_cancel (hr->stall_task);
+ hr->stall_task = NULL;
}
- if (0 == response_code)
+ if (NULL != hr->unpause_task)
{
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Failed to receive response from HTTP server\n");
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_FAILED;
- return;
+ GNUNET_SCHEDULER_cancel (hr->unpause_task);
+ hr->unpause_task = NULL;
}
- /* libgnunetcurl stops buffering at #GNUNET_MAX_MALLOC_CHECKED (40
- MiB) by returning 0 to libcurl, which fails the transfer — but
- CURLINFO_RESPONSE_CODE still holds the status line libcurl parsed
- long before, and GNUNET_CURL_perform2() hands us that code
- together with the partial buffer. Nothing else here distinguishes
- a truncated body from a complete one, and MHD would go on to
- compute a perfectly consistent Content-Length for the fragment:
- the client would receive a corrupt representation under the
- upstream's 2xx with no signal at all. RFC 9112 §8.1.2 requires an
- incomplete message to be treated as a failure, so fail it. The
- same comparison catches an upstream whose declared length and body
- simply disagree, which is the same framing error. */
- if ( (PH_NO_CONTENT_LENGTH != hr->upstream_content_length) &&
- (! hr->head_request) &&
- (status_allows_body ((unsigned int) response_code)) &&
- (hr->upstream_content_length != (uint64_t) body_size) )
- {
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "Upstream declared %llu bytes for `%s' but delivered %llu;"
- " failing the request rather than serving a partial body\n",
- (unsigned long long) hr->upstream_content_length,
- hr->url,
- (unsigned long long) body_size);
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_FAILED;
- return;
- }
- /* A chunked upstream states no length at all, so the comparison
- above cannot see the ceiling being reached. What is still visible
- is how close we came to it: libgnunetcurl refuses the write that
- *would* reach #GNUNET_MAX_MALLOC_CHECKED, and libcurl hands it at
- most CURL_MAX_WRITE_SIZE per call, so every truncated body ends up
- within that much of the ceiling. Bodies that merely happen to
- fall in the same narrow band are refused along with them: at 16 kb
- below a 40 MiB cap that is a fair price for not serving a
- silently-cut representation. */
- if ( ((uint64_t) body_size) + CURL_MAX_WRITE_SIZE >=
- (uint64_t) GNUNET_MAX_MALLOC_CHECKED)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "Response for `%s' reached the %llu byte buffering limit"
- " at %llu bytes; failing the request rather than serving"
- " a partial body\n",
- hr->url,
- (unsigned long long) GNUNET_MAX_MALLOC_CHECKED,
- (unsigned long long) body_size);
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_FAILED;
- return;
- }
- hr->response = MHD_create_response_from_buffer_copy (body_size,
- body);
- if (NULL == hr->response)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Could not allocate memory for response of %llu bytes\n",
- (unsigned long long) body_size);
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_FAILED;
- return;
- }
- hr->own_response = true;
- hr->response_code = response_code;
- /* Restore the upstream's Content-Length where it describes a
- representation rather than the bytes that follow.
- `curl_check_hdr()' drops it unconditionally, which is right for a
- 200 (MHD regenerates it from the buffer) and wrong here: the
- buffer is legitimately empty while the field must still state the
- length the equivalent GET would have had. RFC 9110 §8.6 makes
- that a MUST NOT in both directions -- a HEAD response must not
- carry a length that differs from the GET's, and neither must a
- 304 differ from the 200 it stands in for -- and MHD synthesises
- `Content-Length: 0', which is exactly such a difference.
-
- MHD_RF_HEAD_ONLY_RESPONSE is what makes MHD accept a declared
- length with no body behind it; without it MHD overrides us. */
- if ( (PH_NO_CONTENT_LENGTH != hr->upstream_content_length) &&
- (0 == body_size) /* MHD_RF_HEAD_ONLY_RESPONSE requires it */ &&
- ( (hr->head_request) ||
- (MHD_HTTP_NOT_MODIFIED == (unsigned int) response_code) ) )
- {
- char clbuf[24];
-
- GNUNET_snprintf (clbuf,
- sizeof (clbuf),
- "%llu",
- (unsigned long long) hr->upstream_content_length);
- if (MHD_YES !=
- MHD_set_response_options (hr->response,
- MHD_RF_HEAD_ONLY_RESPONSE,
- MHD_RO_END))
- {
- GNUNET_break (0);
- }
- else if (MHD_YES !=
- MHD_add_response_header (hr->response,
- MHD_HTTP_HEADER_CONTENT_LENGTH,
- clbuf))
- {
- GNUNET_break (0);
- }
- }
- /* RFC 9110 §7.6.1: drop the headers named by the upstream's
- Connection list. Deferred to here (rather than done in
- `curl_check_hdr()') because Connection may arrive after the
- headers it names. */
- {
- struct HttpResponseHeader *nxt;
-
- for (struct HttpResponseHeader *header = hr->header_head;
- NULL != header;
- header = nxt)
- {
- nxt = header->next;
- if (! connection_lists_header (hr->upstream_connection,
- header->type))
- continue;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Dropping connection-specific response header %s\n",
- header->type);
- GNUNET_CONTAINER_DLL_remove (hr->header_head,
- hr->header_tail,
- header);
- GNUNET_free (header->type);
- GNUNET_free (header->value);
- GNUNET_free (header);
- }
- }
- for (struct HttpResponseHeader *header = hr->header_head;
- NULL != header;
- header = header->next)
+ if (UP_DONE != hr->up_state)
+ /* Whatever the client still owes us is no longer of interest, but
+ it has to be read out of the way before MHD will let us queue.
+ See #UP_DRAINING. */
+ hr->up_state = UP_DRAINING;
+ if (0 == response_code)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Adding MHD response header %s->%s\n",
- header->type,
- header->value);
- if (MHD_YES !=
- MHD_add_response_header (hr->response,
- header->type,
- header->value))
- {
- /* MHD validates the field name (RFC 9112 §5 token) and the
- value; a malformed one is the upstream's fault, not ours, so
- GNUNET_break_op and not GNUNET_break -- the latter let an
- eccentric origin fill the log with "Assertion failed". */
- GNUNET_break_op (0);
+ /* The transfer failed. Which of the two failures it was depends
+ on how far we had already got, and only we can tell: a chunked
+ response cut off before its terminating chunk fails here just
+ like an unreachable origin, and libgnunetcurl reports both as
+ zero precisely so that the difference is ours to draw. */
+ if (hr->headers_complete)
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Upstream `%s' sent a header MHD rejected: `%s'\n",
+ "Upstream `%s' failed after %llu bytes of a response"
+ " already in flight; it can only be truncated\n",
hr->url,
- header->type);
- }
+ (unsigned long long) (hr->down_sent
+ + hr->down_ring.len));
+ else
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Failed to receive response from HTTP server\n");
+ hr->down_state = DOWN_FAILED;
+ wake_client (hr);
+ return;
+ }
+ if (! hr->headers_complete)
+ {
+ /* libcurl called the transfer a success without ever ending a
+ header section. Should not happen; nothing has been promised
+ to the client yet, so an ordinary gateway failure is the honest
+ answer. */
+ GNUNET_break (0);
+ hr->down_state = DOWN_FAILED;
+ wake_client (hr);
+ return;
}
- if ( (REQUEST_STATE_PROXY_DOWNLOAD_STARTED == hr->state) ||
- (REQUEST_STATE_PROXY_UPLOAD_STARTED == hr->state) )
+ if ( (PH_NO_CONTENT_LENGTH != hr->upstream_content_length) &&
+ (! hr->head_request) &&
+ (status_allows_body (hr->response_code)) &&
+ (hr->upstream_content_length !=
+ hr->down_sent + (uint64_t) hr->down_ring.len) )
{
- /* Either the request body was empty (CURLOPT_POSTFIELDSIZE = 0,
- so libcurl never called our read callback) or the upload
- already drained but we have not yet entered upload_cb to
- flip the state — either way, the upload is complete and we
- can move straight on to consuming the response. */
- hr->io_len = 0; /* drains upload if it was not yet due to
- early response from server */
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_DONE;
+ /* The origin promised a length and did not deliver it. The
+ promise is already with the client, so the only signal left is
+ to break the framing: see `body_reader()'. */
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Upstream declared %llu bytes for `%s' but delivered %llu;"
+ " the response is already in flight, so it can only be"
+ " truncated\n",
+ (unsigned long long) hr->upstream_content_length,
+ hr->url,
+ (unsigned long long) (hr->down_sent + hr->down_ring.len));
+ hr->down_state = DOWN_FAILED;
+ wake_client (hr);
+ return;
}
+ hr->down_state = DOWN_COMPLETE;
+ wake_client (hr);
}
/**
- * cURL callback for uploaded (PUT/POST) data.
- * Copies from our `io_buf` to make it available to cURL.
+ * cURL callback asking for the next piece of the request body.
*
* @param buf where to write the data
* @param size number of bytes per member
* @param nmemb number of members available in @a buf
* @param cls our `struct HttpRequest` that generated the data
- * @return number of bytes copied to @a buf
+ * @return number of bytes copied to @a buf, 0 for end of body,
+ * #CURL_READFUNC_PAUSE or #CURL_READFUNC_ABORT
*/
static size_t
curl_upload_cb (void *buf,
@@ -1492,63 +2023,180 @@ curl_upload_cb (void *buf,
{
struct HttpRequest *hr = cls;
size_t len = size * nmemb;
- size_t to_copy;
+ size_t n;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Upload cb is working...\n");
-
- if (REQUEST_STATE_PROXY_DOWNLOAD_DONE == hr->state)
+ n = ring_get (&hr->up_ring,
+ buf,
+ len);
+ if (0 != n)
{
- /* Should not happen: the curl task only declares the transfer
- done after consuming POSTFIELDSIZE bytes from us. */
- GNUNET_break (0);
- return CURL_READFUNC_ABORT;
+ note_progress (hr);
+ /* Room freed: MHD may hand us more of the client's body. */
+ wake_client (hr);
+ return n;
}
-
- if (REQUEST_STATE_PROXY_DOWNLOAD_STARTED == hr->state)
+ switch (hr->up_state)
{
- /* We have already drained io_buf and flipped the state in a
- previous call. With CURLOPT_POSTFIELDSIZE set, libcurl
- should not ask for more bytes — but if it does, signal a
- clean end-of-upload rather than aborting the transfer. Both
- conditions describe libcurl's behaviour rather than one of our
- own invariants, so neither may abort the daemon: this process
- is serving every other client too. */
- GNUNET_break (0 == hr->io_len);
+ case UP_HEADERS_PENDING:
+ case UP_STREAMING:
+ /* The client has not caught up. Stop asking until it does. */
+ hr->curl_send_paused = true;
+ return CURL_READFUNC_PAUSE;
+ case UP_CLIENT_DONE:
+ hr->up_state = UP_DONE;
return 0;
+ case UP_DONE:
+ return 0;
+ case UP_DRAINING:
+ /* The client went away, or we gave up on it, in the middle of a
+ body whose length we already declared upstream. We cannot
+ deliver what we promised, so the origin has to be told the
+ request is broken rather than left waiting for bytes that will
+ never come. */
+ return CURL_READFUNC_ABORT;
}
+ GNUNET_break (0);
+ return CURL_READFUNC_ABORT;
+}
+
+
+/**
+ * MHD content reader: hand the client the next piece of the response.
+ *
+ * The contract, stated in our own terms rather than MHD 1.x's so that
+ * the MHD2 port is a rewrite against something written down: return
+ * however many bytes fit in @a buf, or suspend the connection and
+ * return nothing, or declare the stream finished -- with an error if
+ * the body is short of what was promised, cleanly if it is not.
+ *
+ * @param cls our `struct HttpRequest *`
+ * @param pos byte offset in the response (unused; the ring is the
+ * cursor)
+ * @param[out] buf where to write
+ * @param max capacity of @a buf
+ * @return number of bytes written, 0 (suspended),
+ * #MHD_CONTENT_READER_END_OF_STREAM or
+ * #MHD_CONTENT_READER_END_WITH_ERROR
+ */
+static ssize_t
+body_reader (void *cls,
+ uint64_t pos,
+ char *buf,
+ size_t max)
+{
+ struct HttpRequest *hr = cls;
+ size_t n;
- if (REQUEST_STATE_PROXY_UPLOAD_STARTED != hr->state)
+ if (pos != hr->down_sent)
{
- /* Some state that never arms the read callback. Refuse the body
- rather than send whatever io_buf happens to hold. */
+ /* The ring *is* the cursor, so it cannot go back to an offset it
+ has already handed out. MHD 1.x never asks -- it caches a
+ ready block itself and only ever advances, which is why this
+ has held across every case in the suite -- but silently serving
+ the wrong bytes would be far worse than failing the response.
+ Anything that made this fire would also be the first thing to
+ check when porting to MHD2. */
GNUNET_break (0);
- return CURL_READFUNC_ABORT;
+ return MHD_CONTENT_READER_END_WITH_ERROR;
}
- if (0 == hr->io_len)
+ if (hr->mhd_gone)
{
- /* Should not happen: start_curl_request runs only after the
- client body is fully buffered, so io_len begins > 0 (or 0
- for an empty body, which CURLOPT_POSTFIELDSIZE = 0 keeps
- libcurl from asking about). */
+ /* Cannot happen: MHD does not read from a request it has already
+ reported as completed. Answering rather than asserting keeps a
+ surprise here from taking the daemon down. */
GNUNET_break (0);
- return CURL_READFUNC_PAUSE;
+ return MHD_CONTENT_READER_END_WITH_ERROR;
+ }
+ n = ring_get (&hr->down_ring,
+ buf,
+ max);
+ if (0 != n)
+ {
+ hr->down_sent += n;
+ /* Room freed: let the origin send again. */
+ schedule_recv_cont (hr);
+ return (ssize_t) n;
+ }
+ switch (hr->down_state)
+ {
+ case DOWN_COMPLETE:
+ return MHD_CONTENT_READER_END_OF_STREAM;
+ case DOWN_FAILED:
+ /* Never END_OF_STREAM here: under chunked framing that writes the
+ terminating zero-length chunk, which tells the client a
+ truncated body was complete. END_WITH_ERROR closes without it,
+ and closes short of a declared Content-Length, both of which RFC
+ 9112 §8.1.2 requires the client to treat as a failure. */
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Truncating the response to `%s' after %llu bytes\n",
+ hr->url,
+ (unsigned long long) hr->down_sent);
+ return MHD_CONTENT_READER_END_WITH_ERROR;
+ case DOWN_WAITING_HEADERS:
+ case DOWN_STREAMING:
+ /* Nothing to give yet. Returning 0 without suspending would leave
+ MHD in NORMAL_BODY_UNREADY with MHD_EVENT_LOOP_INFO_PROCESS,
+ i.e. spinning. */
+ suspend_client (hr);
+ schedule_recv_cont (hr);
+ return 0;
}
+ GNUNET_break (0);
+ return MHD_CONTENT_READER_END_WITH_ERROR;
+}
- to_copy = GNUNET_MIN (hr->io_len,
- len);
- GNUNET_memcpy (buf,
- &hr->io_buf[hr->io_off],
- to_copy);
- hr->io_off += to_copy;
- hr->io_len -= to_copy;
- if (0 == hr->io_len)
- {
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_STARTED;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Completed CURL UPLOAD\n");
- }
- return to_copy;
+
+/**
+ * Release @a hr once neither MHD nor the content reader can reach it.
+ *
+ * @param[in] hr request to free
+ */
+static void
+reverse_destroy (struct HttpRequest *hr)
+{
+ ring_free (&hr->up_ring);
+ ring_free (&hr->down_ring);
+ free_response_headers (hr);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Proxying of '%s' completely done\n",
+ hr->url);
+ GNUNET_free (hr->url);
+ GNUNET_free (hr->client_via);
+ GNUNET_free (hr->client_xff);
+ GNUNET_free (hr->client_xfp);
+ GNUNET_free (hr->client_xfh);
+ GNUNET_free (hr->client_xfport);
+ GNUNET_free (hr->max_forwards);
+ GNUNET_free (hr->client_forwarded);
+ GNUNET_free (hr->client_connection);
+ GNUNET_free (hr->upstream_connection);
+ GNUNET_CONTAINER_DLL_remove (hr_head,
+ hr_tail,
+ hr);
+ GNUNET_free (hr);
+}
+
+
+/**
+ * MHD is done with the streaming response and is dropping its
+ * reference to our closure.
+ *
+ * The second half of the ownership handshake: MHD calls its completion
+ * notifier -- and therefore `PAIVANA_HTTPD_reverse_cleanup()' -- while
+ * it still holds this response, and destroys the response only
+ * afterwards (see `connection.c' 1344-1354 and 7523-7532). Whichever
+ * of the two runs last does the freeing.
+ *
+ * @param cls our `struct HttpRequest *`
+ */
+static void
+body_reader_free (void *cls)
+{
+ struct HttpRequest *hr = cls;
+
+ hr->reader_gone = true;
+ if (hr->mhd_gone)
+ reverse_destroy (hr);
}
@@ -1840,8 +2488,20 @@ PAIVANA_HTTPD_reverse_create (struct MHD_Connection *connection,
struct HttpRequest *hr;
hr = GNUNET_new (struct HttpRequest);
- hr->state = REQUEST_STATE_HEADERS_PENDING;
+ hr->up_state = UP_HEADERS_PENDING;
+ hr->down_state = DOWN_WAITING_HEADERS;
hr->upstream_content_length = PH_NO_CONTENT_LENGTH;
+ /* No content reader is armed until the streaming response is
+ created, so until then this half of the handshake is already
+ satisfied. */
+ hr->reader_gone = true;
+ /* A ring that cannot take one libcurl delivery would pause with an
+ empty buffer and never be woken; the floor is what rules that
+ out, whatever the operator configured. */
+ hr->up_ring.cap = GNUNET_MAX (PH_RING_MIN_SIZE,
+ PH_request_buffer_max);
+ hr->down_ring.cap = GNUNET_MAX (PH_RING_MIN_SIZE,
+ PH_response_buffer_max);
hr->con = connection;
hr->url = GNUNET_strdup (url);
GNUNET_CONTAINER_DLL_insert (hr_head,
@@ -1854,6 +2514,8 @@ PAIVANA_HTTPD_reverse_create (struct MHD_Connection *connection,
void
PAIVANA_HTTPD_reverse_cleanup (struct HttpRequest *hr)
{
+ struct MHD_Response *resp;
+
if (NULL != hr->curl)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1866,8 +2528,22 @@ PAIVANA_HTTPD_reverse_cleanup (struct HttpRequest *hr)
GNUNET_SCHEDULER_cancel (hr->timeout_task);
hr->timeout_task = NULL;
}
+ if (NULL != hr->stall_task)
+ {
+ GNUNET_SCHEDULER_cancel (hr->stall_task);
+ hr->stall_task = NULL;
+ }
+ if (NULL != hr->unpause_task)
+ {
+ GNUNET_SCHEDULER_cancel (hr->unpause_task);
+ hr->unpause_task = NULL;
+ }
if (NULL != hr->job)
{
+ /* Also the point of cancelling here rather than letting the
+ transfer run out: a download whose client has gone stops
+ consuming the origin's bandwidth instead of being read to the
+ end into a buffer nobody will ever look at. */
GNUNET_CURL_job_cancel (hr->job);
hr->job = NULL;
}
@@ -1876,208 +2552,204 @@ PAIVANA_HTTPD_reverse_cleanup (struct HttpRequest *hr)
curl_slist_free_all (hr->headers);
hr->headers = NULL;
}
- if ( (NULL != hr->response) &&
+ /* The connection is MHD's again the moment we return from its
+ completion notifier, so nothing may reach for it after this. */
+ hr->con = NULL;
+ hr->suspended = GNUNET_NO;
+ resp = hr->response;
+ hr->response = NULL;
+ if ( (NULL != resp) &&
(hr->own_response) )
/* The shared error responses belong to this module, not to the
request that queued one; releasing them here would take a
- reference we never held. */
- MHD_destroy_response (hr->response);
+ reference we never held.
+
+ Dropping our reference may run the content reader's free
+ callback right here, when MHD held no reference of its own --
+ which is why @e mhd_gone is still false at this point. The
+ callback then only records that it has fired, and the check
+ below does the freeing. */
+ MHD_destroy_response (resp);
+ hr->mhd_gone = true;
+ if (hr->reader_gone)
+ reverse_destroy (hr);
+}
- free_response_headers (hr);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Proxying of '%s' completely done\n",
- hr->url);
- GNUNET_free (hr->url);
- GNUNET_free (hr->io_buf);
- GNUNET_free (hr->client_via);
- GNUNET_free (hr->client_xff);
- GNUNET_free (hr->client_xfp);
- GNUNET_free (hr->client_xfh);
- GNUNET_free (hr->client_xfport);
- GNUNET_free (hr->max_forwards);
- GNUNET_free (hr->client_forwarded);
- GNUNET_free (hr->client_connection);
- GNUNET_free (hr->upstream_connection);
- GNUNET_CONTAINER_DLL_remove (hr_head,
- hr_tail,
- hr);
- GNUNET_free (hr);
-}
+/**
+ * How the client framed its request body, if it sent one at all.
+ */
+enum ClientBodyKind
+{
+ /**
+ * No body: neither `Content-Length` nor `Transfer-Encoding`, or a
+ * `Content-Length` of zero.
+ */
+ CLIENT_BODY_NONE,
+
+ /**
+ * A body of a length the client stated.
+ */
+ CLIENT_BODY_LENGTH,
+
+ /**
+ * A body whose length is not known in advance.
+ */
+ CLIENT_BODY_CHUNKED,
+
+ /**
+ * A `Content-Length` we could not make sense of. Not the same as
+ * #CLIENT_BODY_NONE: MHD has already accepted the request, so
+ * something may still arrive.
+ */
+ CLIENT_BODY_BAD
+};
/**
- * Parse the client's Content-Length header (if any) and decide
- * whether the declared body size fits within our per-request
- * buffer cap. Returns false if the header is present, well-formed,
- * and exceeds `PH_request_buffer_max`; the caller must then
- * transition to the reject path to suppress the implicit 100
- * Continue and avoid buffering a body we would only throw away.
+ * Work out whether the client is sending a body, and how it is framed.
+ *
+ * Replaces looking at how much body had been buffered, which is no
+ * longer knowable at the time the origin request is built: it now
+ * starts before the body arrives. Reading the client's own framing is
+ * in any case the more correct test — a `Content-Length: 0` is a
+ * declared empty body, not the absence of one.
*
- * @param con MHD connection to look up the header on
- * @return true if OK to continue accepting the body
+ * @param con MHD connection to look the headers up on
+ * @param[out] len set to the declared length for #CLIENT_BODY_LENGTH
+ * @return how the body is framed
*/
-static bool
-content_length_ok (struct MHD_Connection *con)
+static enum ClientBodyKind
+client_body_kind (struct MHD_Connection *con,
+ uint64_t *len)
{
const char *cl_str;
char *endptr;
unsigned long long cl;
+ *len = 0;
cl_str = MHD_lookup_connection_value (con,
MHD_HEADER_KIND,
MHD_HTTP_HEADER_CONTENT_LENGTH);
if (NULL == cl_str)
- return true;
+ {
+ if (NULL != MHD_lookup_connection_value (con,
+ MHD_HEADER_KIND,
+ MHD_HTTP_HEADER_TRANSFER_ENCODING))
+ return CLIENT_BODY_CHUNKED;
+ return CLIENT_BODY_NONE;
+ }
/* RFC 9110 §8.6: Content-Length = 1*DIGIT. strtoull() is more
generous than that -- it skips leading whitespace and accepts a
sign, so " +5" and even "-1" parse -- and the sign is the one that
matters here, since strtoull() maps a negative value onto a huge
unsigned one. Check the syntax ourselves before believing the
number. */
+ if ('\0' == *cl_str)
+ return CLIENT_BODY_BAD;
for (const char *p = cl_str; '\0' != *p; p++)
if ( ('0' > *p) ||
('9' < *p) )
- return true; /* unparseable — defer judgment to the drain path */
+ return CLIENT_BODY_BAD;
errno = 0;
cl = strtoull (cl_str,
&endptr,
10);
if (endptr == cl_str)
- return true; /* empty field value — likewise not our call */
+ return CLIENT_BODY_BAD;
if (ERANGE == errno)
{
/* All digits, but beyond ULLONG_MAX: larger than any limit we
- could be configured with. Returning true here meant answering
- `Content-Length: 99999999999999999999999' with a 100 Continue
- and then buffering and discarding the body -- precisely the work
- this function exists to skip. */
+ could be configured with, so it is a rejection rather than a
+ parse failure. */
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
"Rejecting upload: Content-Length `%s' is out of range\n",
cl_str);
- return false;
- }
- {
- /* The buffering path also enforces the hard GNUNET_MAX_MALLOC_CHECKED
- (40 MiB) cap, so a Content-Length above that will be rejected
- regardless of how large PH_request_buffer_max is configured. */
- unsigned long long limit =
- GNUNET_MIN (PH_request_buffer_max,
- (unsigned long long) GNUNET_MAX_MALLOC_CHECKED);
-
- if (cl <= limit)
- return true;
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Rejecting upload: Content-Length %llu exceeds %llu byte limit\n",
- cl,
- limit);
- return false;
+ *len = UINT64_MAX;
+ return CLIENT_BODY_LENGTH;
}
+ if (0 == cl)
+ return CLIENT_BODY_NONE;
+ *len = (uint64_t) cl;
+ return CLIENT_BODY_LENGTH;
}
/**
- * Append an upload chunk from the client into `hr->io_buf`, growing
- * the buffer as needed. Returns false when the chunk would push
- * the total past `PH_request_buffer_max`; the caller must then
- * transition to the drain path (MHD disallows queuing a response
- * while `BODY_RECEIVING`, so we silently discard the rest and
- * queue the 413 once MHD calls us back at `FULL_REQ_RECEIVED`).
+ * Take as much of an upload chunk from the client as @e up_ring will
+ * hold, and enforce #PH_max_request_size on the way.
+ *
+ * Unlike libcurl's write callback, MHD's access handler may take part
+ * of what it is given: it panics only if the handler claims *more*
+ * than it was handed, and warns only when the handler takes nothing
+ * and does not suspend (`connection.c' 4774-4792). So the ring-full
+ * case has to suspend, and does.
*
* @param[in,out] hr request we are handling
- * @param upload_data_size number of bytes in @a upload_data
- * @param upload_data data being uploaded
- * @return true on success, false if the upload is too big
+ * @param[in,out] upload_data_size in: bytes offered, out: bytes left
+ * @param upload_data the bytes
+ * @return true if the upload may continue, false if it is now over
+ * #PH_max_request_size and the 413 path has to take over
*/
static bool
-buffer_upload_chunk (struct HttpRequest *hr,
- size_t upload_data_size,
- const char upload_data[static upload_data_size])
+relay_upload_chunk (struct HttpRequest *hr,
+ size_t *upload_data_size,
+ const char *upload_data)
{
- uint64_t min_size;
+ size_t n;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Processing %u bytes UPLOAD\n",
- (unsigned int) upload_data_size);
- min_size = ((uint64_t) hr->io_len) + ((uint64_t) upload_data_size);
- if ( (min_size < hr->io_len /* integer overflow */) ||
- (min_size > PH_request_buffer_max /* overflows config limit */) ||
- (min_size > GNUNET_MAX_MALLOC_CHECKED) ||
- (min_size > UINT_MAX /* would overflow later */) )
+ if (hr->up_received + (uint64_t) *upload_data_size > PH_max_request_size)
{
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Upload exceeds limit, rejecting\n");
+ "Rejecting upload: body exceeds the %llu byte limit\n",
+ (unsigned long long) PH_max_request_size);
return false;
}
- if (hr->io_size - hr->io_len < upload_data_size)
- {
- unsigned int new_size;
- char *nbuf;
-
- new_size = GNUNET_MIN (
- GNUNET_MAX_MALLOC_CHECKED,
- GNUNET_MAX (
- GNUNET_MIN (hr->io_size * 2 + 1024,
- UINT_MAX),
- min_size));
- GNUNET_assert (new_size > hr->io_size);
- /* Deliberately not GNUNET_array_grow(): that allocates through
- GNUNET_xmalloc_(), which calls GNUNET_assert(0) when malloc
- returns NULL, i.e. aborts the whole daemon -- dropping every
- other client, including the paying ones -- because one request
- could not be buffered. Degrade to 413 for this request instead
- and keep serving. */
- nbuf = GNUNET_malloc_large (new_size);
- if (NULL == nbuf)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Could not allocate %u bytes to buffer an upload;"
- " rejecting this request rather than terminating\n",
- new_size);
- return false;
- }
- GNUNET_memcpy (nbuf,
- hr->io_buf,
- hr->io_len);
- GNUNET_free (hr->io_buf);
- hr->io_buf = nbuf;
- hr->io_size = new_size;
- }
- GNUNET_memcpy (&hr->io_buf[hr->io_len],
- upload_data,
- upload_data_size);
- hr->io_len += upload_data_size;
+ n = ring_put (&hr->up_ring,
+ upload_data,
+ *upload_data_size,
+ false);
+ hr->up_received += n;
+ *upload_data_size -= n;
+ if (0 != n)
+ {
+ note_progress (hr);
+ /* The origin may have run out of body to send; tell it there is
+ more. */
+ schedule_send_cont (hr);
+ }
return true;
}
/**
- * Arrange for a buffered request body to be forwarded under @a meth,
- * and set @a hr->state to match.
+ * Arrange for a request body to be forwarded under @a meth, and set
+ * @a hr->up_state to match.
*
* For the methods whose curl option is otherwise bodyless
* (CURLOPT_HTTPGET, CURLOPT_NOBODY, or a bare CURLOPT_CUSTOMREQUEST),
* turning on CURLOPT_POST is what enables the read callback;
* CURLOPT_CUSTOMREQUEST then restores the verb on the wire. This is
- * the same shape DELETE has always used. With no body buffered
+ * the same shape DELETE has always used. With no body announced
* nothing is changed, so the common case still goes out as a plain
* GET.
*
* @param[in,out] hr the request
* @param meth HTTP method to keep on the wire
+ * @param kind how the client framed its body
*/
static void
forward_body_verbatim (struct HttpRequest *hr,
- const char *meth)
+ const char *meth,
+ enum ClientBodyKind kind)
{
- if (0 == hr->io_len)
+ if (CLIENT_BODY_NONE == kind)
{
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_STARTED;
+ hr->up_state = UP_DONE;
return;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Forwarding a %llu byte request body with %s\n",
- (unsigned long long) hr->io_len,
+ "Forwarding a request body with %s\n",
meth);
PH_SETOPT (hr->curl,
CURLOPT_POST,
@@ -2085,29 +2757,30 @@ forward_body_verbatim (struct HttpRequest *hr,
PH_SETOPT (hr->curl,
CURLOPT_CUSTOMREQUEST,
meth);
- hr->state = REQUEST_STATE_PROXY_UPLOAD_STARTED;
+ hr->up_state = UP_STREAMING;
}
/**
- * Choose the curl options for the HTTP method we're proxying and
- * set the next proxy state accordingly. Queues an error response
- * and returns the corresponding MHD_Result for unsupported methods;
- * otherwise returns #MHD_YES and leaves @a hr->state advanced to
- * either #PROXY_UPLOAD_STARTED (method has a body to forward) or
- * #PROXY_DOWNLOAD_STARTED (bodyless request).
+ * Choose the curl options for the HTTP method we're proxying and set
+ * @a hr->up_state accordingly. Queues an error response and returns
+ * the corresponding MHD_Result for unsupported methods; otherwise
+ * returns #MHD_YES with @a hr->up_state at #UP_STREAMING (there is a
+ * body to relay) or #UP_DONE (there is not).
*
* On error, the "curl" handle is set to NULL (!).
*
* @param[in,out] hr the request
* @param con client connection handle from MHD
* @param meth HTTP method specified by the client
+ * @param kind how the client framed its request body
* @return MHD status to return
*/
static enum MHD_Result
configure_curl_method (struct HttpRequest *hr,
struct MHD_Connection *con,
- const char *meth)
+ const char *meth,
+ enum ClientBodyKind kind)
{
if (0 == strcasecmp (meth,
MHD_HTTP_METHOD_GET))
@@ -2123,7 +2796,8 @@ configure_curl_method (struct HttpRequest *hr,
arrived, and the client with a parse error it could not
explain. */
forward_body_verbatim (hr,
- MHD_HTTP_METHOD_GET);
+ MHD_HTTP_METHOD_GET,
+ kind);
return MHD_YES;
}
if (0 == strcasecmp (meth,
@@ -2134,7 +2808,7 @@ configure_curl_method (struct HttpRequest *hr,
PH_SETOPT (hr->curl,
CURLOPT_POST,
1L);
- hr->state = REQUEST_STATE_PROXY_UPLOAD_STARTED;
+ hr->up_state = UP_STREAMING;
return MHD_YES;
}
if (0 == strcasecmp (meth,
@@ -2149,7 +2823,8 @@ configure_curl_method (struct HttpRequest *hr,
/* HEAD must be forwarded with whatever body the equivalent GET
carried, or the origin cannot answer the question HEAD asks. */
forward_body_verbatim (hr,
- MHD_HTTP_METHOD_HEAD);
+ MHD_HTTP_METHOD_HEAD,
+ kind);
return MHD_YES;
}
if (0 == strcasecmp (meth,
@@ -2160,7 +2835,7 @@ configure_curl_method (struct HttpRequest *hr,
PH_SETOPT (hr->curl,
CURLOPT_UPLOAD,
1L);
- hr->state = REQUEST_STATE_PROXY_UPLOAD_STARTED;
+ hr->up_state = UP_STREAMING;
return MHD_YES;
}
if (0 == strcasecmp (meth,
@@ -2171,17 +2846,17 @@ configure_curl_method (struct HttpRequest *hr,
PH_SETOPT (hr->curl,
CURLOPT_CUSTOMREQUEST,
"DELETE");
- if (0 != hr->io_len)
+ if (CLIENT_BODY_NONE != kind)
{
/* DELETE with a request body is unusual but legal. */
PH_SETOPT (hr->curl,
CURLOPT_POST,
1L);
- hr->state = REQUEST_STATE_PROXY_UPLOAD_STARTED;
+ hr->up_state = UP_STREAMING;
}
else
{
- hr->state = REQUEST_STATE_PROXY_DOWNLOAD_STARTED;
+ hr->up_state = UP_DONE;
}
return MHD_YES;
}
@@ -2198,7 +2873,7 @@ configure_curl_method (struct HttpRequest *hr,
PH_SETOPT (hr->curl,
CURLOPT_CUSTOMREQUEST,
"PATCH");
- hr->state = REQUEST_STATE_PROXY_UPLOAD_STARTED;
+ hr->up_state = UP_STREAMING;
return MHD_YES;
}
if (0 == strcasecmp (meth,
@@ -2259,7 +2934,8 @@ configure_curl_method (struct HttpRequest *hr,
hops - 1);
}
forward_body_verbatim (hr,
- MHD_HTTP_METHOD_OPTIONS);
+ MHD_HTTP_METHOD_OPTIONS,
+ kind);
return MHD_YES;
}
/* TRACE leaks headers back to the client; CONNECT is for TLS
@@ -2646,11 +3322,15 @@ append_forwarded_headers (struct HttpRequest *hr,
/**
- * Initialize the curl handle, attach the forwarding headers, and
- * hand the request off to the curl multi loop. On success
- * advances @a hr->state to one of the PROXY_*_STARTED states and
- * returns #MHD_YES. On any failure queues an appropriate error
- * response and returns its MHD_Result.
+ * Initialize the curl handle, attach the forwarding headers, and hand
+ * the request off to the curl multi loop. On success sets
+ * @a hr->up_state to #UP_STREAMING or #UP_DONE and returns #MHD_YES.
+ * On any failure queues an appropriate error response and returns its
+ * MHD_Result.
+ *
+ * Called before the client's body has arrived -- that is the point of
+ * streaming it -- so what the body will be is read out of the client's
+ * own framing rather than out of a buffer.
*
* On error, the "curl" handle is set to NULL (!).
*
@@ -2658,13 +3338,17 @@ append_forwarded_headers (struct HttpRequest *hr,
* @param con MHD connection handle
* @param meth HTTP method of the request
* @param ver HTTP version to use
+ * @param kind how the client framed its request body
+ * @param body_len declared body length, for #CLIENT_BODY_LENGTH
* @return MHD status code to return
*/
static enum MHD_Result
start_curl_request (struct HttpRequest *hr,
struct MHD_Connection *con,
const char *meth,
- const char *ver)
+ const char *ver,
+ enum ClientBodyKind kind,
+ uint64_t body_len)
{
enum MHD_Result r;
@@ -2674,55 +3358,67 @@ start_curl_request (struct HttpRequest *hr,
if (NULL == hr->curl)
{
PAIVANA_LOG_ERROR ("Could not init the curl handle\n");
- hr->suspended = GNUNET_NO;
- MHD_resume_connection (hr->con);
return MHD_queue_response (con,
MHD_HTTP_INTERNAL_SERVER_ERROR,
internal_failure_response);
}
- /* No need to check whether we're POSTing or PUTting.
- * If not needed, one of the following values will be
- * ignored.*/
- /* The casts are load-bearing, not decoration. Both options are
+ /* The client's own framing is reproduced upstream: a declared length
+ stays a declared length, and a chunked body stays chunked (libcurl
+ chunks for us when the size is left unknown). Both options are
+ set because which of the two libcurl reads depends on whether
+ `configure_curl_method()' chose CURLOPT_POST or CURLOPT_UPLOAD.
+
+ The casts are load-bearing, not decoration. Both options are
CURLOPTTYPE_OFF_T (<curl/curl.h>), so libcurl reads them with
va_arg(curl_off_t), and curl_off_t is 64 bits on every platform
(<curl/system.h>: "For any given platform/compiler curl_off_t MUST
be typedef'ed to a 64-bit wide signed integral data type").
Passing a size_t is therefore only correct where size_t is 64 bits
- too: on i386 or armv7 libcurl takes the four bytes of io_len and
- four bytes of whatever the caller left on the stack next to them
- as the high half, and declares a request body of some absurd
+ too: on i386 or armv7 libcurl would take four bytes of the length
+ and four bytes of whatever the caller left on the stack next to
+ them as the high half, and declare a request body of some absurd
length that never arrives. */
- PH_SETOPT (hr->curl,
- CURLOPT_POSTFIELDSIZE_LARGE,
- (curl_off_t) hr->io_len);
- PH_SETOPT (hr->curl,
- CURLOPT_INFILESIZE_LARGE,
- (curl_off_t) hr->io_len);
- PH_SETOPT (hr->curl,
- CURLOPT_HEADERFUNCTION,
- &curl_check_hdr);
- PH_SETOPT (hr->curl,
- CURLOPT_HEADERDATA,
- hr);
+ if (CLIENT_BODY_LENGTH == kind)
+ {
+ PH_SETOPT (hr->curl,
+ CURLOPT_POSTFIELDSIZE_LARGE,
+ (curl_off_t) body_len);
+ PH_SETOPT (hr->curl,
+ CURLOPT_INFILESIZE_LARGE,
+ (curl_off_t) body_len);
+ }
+ else
+ {
+ PH_SETOPT (hr->curl,
+ CURLOPT_POSTFIELDSIZE_LARGE,
+ (curl_off_t) -1);
+ PH_SETOPT (hr->curl,
+ CURLOPT_INFILESIZE_LARGE,
+ (curl_off_t) -1);
+ }
PH_SETOPT (hr->curl,
CURLOPT_FOLLOWLOCATION,
0L); /* CURLOPTTYPE_LONG: read with va_arg(long) */
PH_SETOPT (hr->curl,
CURLOPT_CONNECTTIMEOUT,
PH_CURL_CONNECT_TIMEOUT_S);
- PH_SETOPT (hr->curl,
- CURLOPT_TIMEOUT,
- PH_CURL_TIMEOUT_S);
+ /* Deliberately no CURLOPT_TIMEOUT and no CURLOPT_LOW_SPEED_LIMIT /
+ _TIME. The first is a deadline on the whole transfer, and a
+ 500 MiB download legitimately outlives any value that would be
+ useful for anything else. The second looks like exactly the
+ watchdog we want and is not; see `stall_check()' for the measured
+ reason, which is that it would kill a slow client's own upload. */
PH_SETOPT (hr->curl,
CURLOPT_NOSIGNAL,
1L);
- /* No CURLOPT_PRIVATE here: GNUNET_CURL_job_add_raw() sets it to its
- own `struct GNUNET_CURL_Job *' and reads it back with
+ /* No CURLOPT_PRIVATE here: GNUNET_CURL_job_add_stream() sets it to
+ its own `struct GNUNET_CURL_Job *' and reads it back with
GNUNET_assert() on the result, so anything we stored would be
overwritten before the transfer starts -- and would abort the
- daemon if it were not. */
+ daemon if it were not. Nor CURLOPT_WRITEFUNCTION or
+ CURLOPT_HEADERFUNCTION, which the streaming job installs its own
+ trampolines into; CURLOPT_READFUNCTION it leaves to us. */
PH_SETOPT (hr->curl,
CURLOPT_VERBOSE,
0L); /* CURLOPTTYPE_LONG: read with va_arg(long) */
@@ -2775,13 +3471,10 @@ start_curl_request (struct HttpRequest *hr,
r = configure_curl_method (hr,
con,
- meth);
+ meth,
+ kind);
if (NULL == hr->curl)
- {
- hr->suspended = GNUNET_NO;
- MHD_resume_connection (hr->con);
return r; /* unsupported method: response already queued */
- }
/* First pass: collect Via / Connection so `con_val_iter` can
honor them (append to Via; drop headers named by Connection). */
MHD_get_connection_values (con,
@@ -2795,22 +3488,31 @@ start_curl_request (struct HttpRequest *hr,
append_forwarded_headers (hr,
con,
ver);
- hr->job = GNUNET_CURL_job_add_raw (PH_proxy_ctx,
- hr->curl,
- hr->headers,
- &curl_download_cb,
- hr);
+ {
+ struct GNUNET_CURL_StreamHandlers sh = {
+ .hcb = &curl_check_hdr,
+ .hcb_cls = hr,
+ .scb = &stream_body_cb,
+ .scb_cls = hr,
+ .jcc = &curl_done_cb,
+ .jcc_cls = hr
+ };
+
+ hr->job = GNUNET_CURL_job_add_stream (PH_proxy_ctx,
+ hr->curl,
+ hr->headers,
+ &sh);
+ }
hr->curl = NULL;
if (NULL == hr->job)
{
GNUNET_break (0);
- hr->suspended = GNUNET_NO;
- MHD_resume_connection (hr->con);
return MHD_queue_response (con,
MHD_HTTP_BAD_GATEWAY,
curl_failure_response);
}
- hr->timeout_task = GNUNET_SCHEDULER_add_delayed (PH_UPSTREAM_TIMEOUT,
+ hr->last_progress = GNUNET_TIME_absolute_get ();
+ hr->timeout_task = GNUNET_SCHEDULER_add_delayed (PH_upstream_timeout,
&upstream_timeout,
hr);
return MHD_YES;
@@ -2818,27 +3520,102 @@ start_curl_request (struct HttpRequest *hr,
/**
- * Queue the response `curl_download_cb()' built for @a hr.
+ * Build and queue the streaming response for @a hr.
+ *
+ * Reached once the origin's header section has ended *and* the
+ * client's request body is out of the way; MHD will not accept a
+ * response before the latter. The body itself has not arrived and
+ * need not: `body_reader()' supplies it as it does.
+ *
+ * The size handed to MHD is what decides the framing the client sees,
+ * and it is the origin's own: a declared `Content-Length` is relayed
+ * as one, and an unknown length becomes chunked for an HTTP/1.1 client
+ * and connection-delimited for an HTTP/1.0 one. Both are what the
+ * origin chose, which is a conformance gain over recomputing a length
+ * from an assembled buffer. It is also what makes a HEAD or a 304
+ * report the length of the body it does not send: MHD skips the
+ * content reader for those but still emits the size (RFC 9110 §9.3.2,
+ * §8.6).
*
* @param[in,out] hr request whose response to queue
* @param con MHD client connection to send the response on
* @return the #MHD_Result of queuing it
*/
static enum MHD_Result
-finalize_response (struct HttpRequest *hr,
- struct MHD_Connection *con)
+queue_stream_response (struct HttpRequest *hr,
+ struct MHD_Connection *con)
{
- /* Reaching #REQUEST_STATE_PROXY_DOWNLOAD_DONE means
- `curl_download_cb()' got far enough to build a response; every
- path on which it did not sets #REQUEST_STATE_PROXY_DOWNLOAD_FAILED
- instead, which is answered elsewhere. A NULL here is thus a bug
- in the state machine, not an upstream failure -- but it is one the
- client should not have to see as a hung connection. */
+ uint64_t size = MHD_SIZE_UNKNOWN;
+
+ if (PH_NO_CONTENT_LENGTH != hr->upstream_content_length)
+ size = hr->upstream_content_length;
+ hr->response = MHD_create_response_from_callback (size,
+ PH_STREAM_BLOCK_SIZE,
+ &body_reader,
+ hr,
+ &body_reader_free);
if (NULL == hr->response)
{
GNUNET_break (0);
hr->response_code = MHD_HTTP_INTERNAL_SERVER_ERROR;
hr->response = internal_failure_response;
+ return MHD_queue_response (con,
+ hr->response_code,
+ hr->response);
+ }
+ hr->own_response = true;
+ /* From here MHD holds a closure that outlives its completion
+ notifier; see @e mhd_gone. */
+ hr->reader_gone = false;
+ /* RFC 9110 §7.6.1: drop the headers named by the upstream's
+ Connection list. Deferred to here (rather than done in
+ `curl_check_hdr()') because Connection may arrive after the
+ headers it names. */
+ {
+ struct HttpResponseHeader *nxt;
+
+ for (struct HttpResponseHeader *header = hr->header_head;
+ NULL != header;
+ header = nxt)
+ {
+ nxt = header->next;
+ if (! connection_lists_header (hr->upstream_connection,
+ header->type))
+ continue;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Dropping connection-specific response header %s\n",
+ header->type);
+ GNUNET_CONTAINER_DLL_remove (hr->header_head,
+ hr->header_tail,
+ header);
+ GNUNET_free (header->type);
+ GNUNET_free (header->value);
+ GNUNET_free (header);
+ }
+ }
+ for (struct HttpResponseHeader *header = hr->header_head;
+ NULL != header;
+ header = header->next)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Adding MHD response header %s->%s\n",
+ header->type,
+ header->value);
+ if (MHD_YES !=
+ MHD_add_response_header (hr->response,
+ header->type,
+ header->value))
+ {
+ /* MHD validates the field name (RFC 9112 §5 token) and the
+ value; a malformed one is the upstream's fault, not ours, so
+ GNUNET_break_op and not GNUNET_break -- the latter let an
+ eccentric origin fill the log with "Assertion failed". */
+ GNUNET_break_op (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Upstream `%s' sent a header MHD rejected: `%s'\n",
+ hr->url,
+ header->type);
+ }
}
return MHD_queue_response (con,
hr->response_code,
@@ -2849,10 +3626,12 @@ finalize_response (struct HttpRequest *hr,
/**
* Main MHD callback for reverse proxy.
*
- * Pure state-machine dispatch: each invocation picks up @a hr->state,
- * runs the transitions it can without new input from MHD, and
- * either returns to wait for more MHD data / curl progress, or
- * cascades through the `while` loop to the next applicable state.
+ * Two independent state machines, not one: since the origin request
+ * now starts before the client's body has arrived, the two directions
+ * run at the same time and neither is a phase of the other. This
+ * function owns the upload half; `stream_body_cb()' and
+ * `body_reader()' own the download half between them, and meet here
+ * only to decide when the response may be queued.
*
* @param hr the HTTP request context
* @param con MHD connection handle
@@ -2879,102 +3658,129 @@ PAIVANA_HTTPD_reverse (struct HttpRequest *hr,
{
(void) url;
- while (true)
+ if (UP_HEADERS_PENDING == hr->up_state)
{
- switch (hr->state)
- {
- case REQUEST_STATE_HEADERS_PENDING:
- /* MHD's HEADERS_PROCESSED callback: we can still queue a
- response here before the client body is consumed, so this
- is our only chance to short-circuit an oversized upload
- based on Content-Length and suppress the implicit 100
- Continue (RFC 7231 §5.1.1). */
- if (! content_length_ok (con))
- {
- hr->state = REQUEST_STATE_REJECT_UPLOAD;
- continue;
- }
- hr->state = REQUEST_STATE_CLIENT_UPLOAD_STARTED;
- return MHD_YES;
-
- case REQUEST_STATE_CLIENT_UPLOAD_STARTED:
- if (0 == *upload_data_size)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Finished processing UPLOAD\n");
- hr->state = REQUEST_STATE_CLIENT_UPLOAD_DONE;
- continue;
- }
- if (buffer_upload_chunk (hr,
- *upload_data_size,
- upload_data))
- {
- *upload_data_size = 0;
- return MHD_YES;
- }
- hr->state = REQUEST_STATE_REJECT_UPLOAD_DRAIN;
- continue;
-
- case REQUEST_STATE_REJECT_UPLOAD_DRAIN:
- /* MHD disallows queuing a response while BODY_RECEIVING, so
- silently discard the remaining body bytes. Once MHD calls
- us with `upload_data_size == 0` (FULL_REQ_RECEIVED) we can
- transition to REJECT_UPLOAD and queue the deferred 413. */
- if (0 != *upload_data_size)
- {
- *upload_data_size = 0;
- return MHD_YES;
- }
- hr->state = REQUEST_STATE_REJECT_UPLOAD;
- continue;
+ enum ClientBodyKind kind;
+ uint64_t body_len;
- case REQUEST_STATE_REJECT_UPLOAD:
+ /* MHD's HEADERS_PROCESSED callback. This is our only chance to
+ refuse an upload before MHD auto-generates a 100 Continue and
+ the client starts sending (RFC 7231 §5.1.1), so a declared
+ length over the limit is answered here and now. */
+ kind = client_body_kind (con,
+ &body_len);
+ if ( (CLIENT_BODY_LENGTH == kind) &&
+ (body_len > PH_max_request_size) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Rejecting upload: Content-Length %llu exceeds the"
+ " %llu byte limit\n",
+ (unsigned long long) body_len,
+ (unsigned long long) PH_max_request_size);
+ hr->reject_upload = true;
+ hr->up_state = UP_DRAINING;
return MHD_queue_response (con,
MHD_HTTP_CONTENT_TOO_LARGE,
upload_failure_response);
+ }
+ return start_curl_request (hr,
+ con,
+ meth,
+ ver,
+ kind,
+ body_len);
+ }
- case REQUEST_STATE_CLIENT_UPLOAD_DONE:
- /* start_curl_request advances state to PROXY_UPLOAD_STARTED
- or PROXY_DOWNLOAD_STARTED on success and returns MHD_YES;
- on error it queues a response and returns its MHD_Result. */
- GNUNET_assert (GNUNET_NO == hr->suspended);
- MHD_suspend_connection (con);
- hr->suspended = GNUNET_YES;
- return start_curl_request (hr,
- con,
- meth,
- ver);
-
- case REQUEST_STATE_PROXY_UPLOAD_STARTED:
- case REQUEST_STATE_PROXY_DOWNLOAD_STARTED:
- /* we should not have been resumed in this state,
- how did we get here? */
- GNUNET_break (0);
- /* Whether MHD can call us here at all is MHD's business, not an
- invariant of ours, so it must not abort the daemon. What we
- do have to preserve is that @e suspended counts the one
- suspend reference we hold: suspending a second time would
- leave a reference nothing ever resumes, and the connection
- would hang until the client gave up. */
- if (GNUNET_NO == hr->suspended)
+ if (0 != *upload_data_size)
+ {
+ /* The client is sending its body. */
+ if (hr->reject_upload ||
+ (UP_DRAINING == hr->up_state))
+ {
+ /* Nobody wants these bytes -- we refused the upload, or the
+ origin finished without them -- but MHD will not let us queue
+ a response while the body is still arriving, so they have to
+ be read and dropped. See #UP_DRAINING. */
+ *upload_data_size = 0;
+ return MHD_YES;
+ }
+ if (! relay_upload_chunk (hr,
+ upload_data_size,
+ upload_data))
+ {
+ /* Over #PH_max_request_size with no Content-Length to have
+ caught it earlier. The origin is told the request is broken;
+ the client gets the 413 once it stops talking. */
+ hr->reject_upload = true;
+ hr->up_state = UP_DRAINING;
+ if (NULL != hr->job)
{
- MHD_suspend_connection (con);
- hr->suspended = GNUNET_YES;
+ GNUNET_CURL_job_cancel (hr->job);
+ hr->job = NULL;
}
+ *upload_data_size = 0;
return MHD_YES;
-
- case REQUEST_STATE_PROXY_DOWNLOAD_DONE:
- return finalize_response (hr,
- con);
- case REQUEST_STATE_PROXY_DOWNLOAD_FAILED:
- return MHD_queue_response (con,
- hr->upstream_timed_out
- ? MHD_HTTP_GATEWAY_TIMEOUT
- : MHD_HTTP_BAD_GATEWAY,
- hr->upstream_timed_out
- ? timeout_failure_response
- : curl_failure_response);
}
- GNUNET_assert (0); /* unreachable */
+ if (0 != *upload_data_size)
+ /* @e up_ring is full: what is left will be offered again once
+ the origin has taken some of it (`curl_upload_cb()' resumes
+ us). Suspending is not optional -- MHD warns about a handler
+ that takes nothing and stays runnable, and spins. */
+ suspend_client (hr);
+ return MHD_YES;
}
+
+ /* `*upload_data_size == 0': the client's request body, if it had
+ one, is complete. MHD will accept a response from here on. */
+ if (UP_STREAMING == hr->up_state)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Finished receiving %llu bytes of request body\n",
+ (unsigned long long) hr->up_received);
+ hr->up_state = UP_CLIENT_DONE;
+ /* The read callback may be parked waiting for bytes that will now
+ never come; let it find that out and report end-of-body. */
+ schedule_send_cont (hr);
+ }
+ if (hr->reject_upload)
+ return MHD_queue_response (con,
+ MHD_HTTP_CONTENT_TOO_LARGE,
+ upload_failure_response);
+ switch (hr->down_state)
+ {
+ case DOWN_WAITING_HEADERS:
+ /* No status to send yet. `curl_check_hdr()' or `curl_done_cb()'
+ will wake us. */
+ suspend_client (hr);
+ return MHD_YES;
+ case DOWN_STREAMING:
+ case DOWN_COMPLETE:
+ return queue_stream_response (hr,
+ con);
+ case DOWN_FAILED:
+ if (hr->headers_complete)
+ {
+ /* The status went out with the headers, so there is nothing
+ left to say: the response can only be truncated. Queue it
+ anyway -- `body_reader()' ends it with an error immediately,
+ which breaks the framing where a bare close would have looked
+ like a complete message. */
+ return queue_stream_response (hr,
+ con);
+ }
+ return MHD_queue_response (con,
+ hr->upstream_timed_out
+ ? MHD_HTTP_GATEWAY_TIMEOUT
+ : MHD_HTTP_BAD_GATEWAY,
+ hr->upstream_timed_out
+ ? timeout_failure_response
+ : curl_failure_response);
+ }
+ GNUNET_break (0);
+ return MHD_queue_response (con,
+ MHD_HTTP_INTERNAL_SERVER_ERROR,
+ internal_failure_response);
}
+
+
+/* end of paivana-httpd_reverse.c */
diff --git a/src/frontend/paywall.js b/src/frontend/paywall.js
@@ -80,14 +80,43 @@ function waitMs(ms) {
}
/**
+ * Encode @a bytes in the RFC 4648 section 5 URL-safe alphabet, with
+ * the padding stripped -- the same form the daemon produces with
+ * GNUNET_STRINGS_base64url_encode(), and the mirror of the decode at
+ * the top of this file.
+ *
+ * Deliberately built out of btoa() rather than the much tidier
+ * `bytes.toBase64({alphabet: "base64url"})'. That method is a 2024-25
+ * addition (Firefox 133, Safari 18.2, Chrome 140), so on anything
+ * older it is simply not a function -- and this runs on the payment
+ * path, where the exception would leave a paywall that cannot be paid
+ * with no indication why. That is the same failure the decode above
+ * exists to prevent, and it is not worth reintroducing at the other
+ * end for the sake of one line.
+ *
+ * @param {Uint8Array} bytes
+ * @returns {string}
+ */
+function base64url(bytes) {
+ let s = "";
+
+ /* Not String.fromCharCode(...bytes): spreading a large array
+ overflows the argument limit. The digests here are 32 bytes, but
+ the loop costs nothing and does not have to be re-examined if a
+ caller ever passes something bigger. */
+ for (const b of bytes) {
+ s += String.fromCharCode(b);
+ }
+ return btoa(s).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
+}
+
+/**
* @param {BufferSource} data
* @returns {Promise<string>}
*/
async function sha256b64(data) {
const buf = await crypto.subtle.digest("SHA-256", data);
- return new Uint8Array(buf)
- .toBase64({ alphabet: "base64url" })
- .replace(/=+$/, "");
+ return base64url(new Uint8Array(buf));
}
/**
diff --git a/src/frontend/typecheck.sh b/src/frontend/typecheck.sh
@@ -8,16 +8,26 @@
# by hand without installing anything:
#
# npx --package typescript tsc --allowJs --checkJs --noEmit \
-# --target esnext --lib esnext,esnext.bigint,dom \
+# --target esnext --lib esnext,esnext.bigint,dom,dom.iterable \
# src/frontend/paywall.js src/frontend/global.d.ts
+#
+# dom.iterable is not optional: without it `for (const x of
+# el.getElementsByClassName(...))' is an error, because plain `dom'
+# declares HTMLCollectionOf without a [Symbol.iterator].
set -eu
srcdir=${SRCDIR:-$(dirname "$0")}
+# `tsc' on npm is NOT the TypeScript compiler -- it is a squatted
+# package whose entire content is a message telling you so, and it
+# exits 0 for --version. A bare `npx --no-install tsc --version'
+# probe therefore succeeds against it and we then run it for real,
+# which fails the test with that message and nothing to do with this
+# code. Insist the probe looks like a version.
if command -v tsc >/dev/null 2>&1; then
set -- tsc
-elif npx --no-install tsc --version >/dev/null 2>&1; then
+elif npx --no-install tsc --version 2>/dev/null | grep -Eq '[0-9]+\.[0-9]+'; then
set -- npx --no-install tsc
else
echo "no local TypeScript compiler found, skipping type check" >&2
@@ -29,6 +39,6 @@ exec "$@" \
--checkJs \
--noEmit \
--target esnext \
- --lib esnext,esnext.bigint,dom \
+ --lib esnext,esnext.bigint,dom,dom.iterable \
"$srcdir/paywall.js" \
"$srcdir/global.d.ts"
diff --git a/src/tests/README b/src/tests/README
@@ -35,7 +35,7 @@ so that paivana is not exercised only against libmicrohttpd peers:
upstream_py Python (stdlib) (pure interpreter; needs python3
at `make check` time)
-A fifth, special-purpose upstream is also built:
+Two further, special-purpose upstreams are also built:
early_response_upstream
C / raw sockets, single-connection. Sends a 413
@@ -58,11 +58,37 @@ A fifth, special-purpose upstream is also built:
deadlock; without `--no-drain` the condition never
arises, because the upstream keeps reading.
-They all implement the same canned endpoints (see "Endpoints" below).
-The pipelining test client `pipeline_client` is a small C program
-that talks directly to the paivana listen socket using BSD sockets.
-
-All five bind 127.0.0.1 and nothing else. They are not hardened in
+ stream_upstream
+ C / raw sockets, one process per connection. Serves
+ bodies far larger than memory, at a rate the driver
+ chooses, and in framings a conforming server library
+ will not emit: a declared `Content-Length` that is
+ not delivered, a chunked response with no terminating
+ chunk, a connection that answers and then goes silent
+ for ever. Also reads and verifies a request body,
+ optionally answering before it has finished.
+
+ Bodies are a deterministic function of their own byte
+ offset rather than stored data, so a 200 MiB case
+ costs no disk on either side. Deliberately not a
+ constant byte: a repeated character would pass a
+ comparison that duplicated or dropped a whole aligned
+ block, which is exactly the mistake a ring buffer
+ with wrong wrap arithmetic makes -- and exactly the
+ mistake this caught during development.
+
+The four canned upstreams all implement the same endpoints (see
+"Endpoints" below). Two test clients talk to paivana directly:
+`pipeline_client`, which uses BSD sockets to pipeline requests, and
+`stream_client`, which verifies a body against the same generated
+pattern as it arrives and reports what it saw about the *framing* --
+whether the response was chunked, what `Content-Length` reached the
+client, how long the first byte took relative to the last. Verifying
+incrementally is the point: a body written to a file and compared
+afterwards says nothing about whether paivana streamed it or assembled
+it first.
+
+They all bind 127.0.0.1 and nothing else. They are not hardened in
any way -- POST /echo reflects whatever body it is given and GET
/large/10485760 hands out 10 MiB per request -- and they have no
business being reachable from the network for the duration of `make
@@ -248,7 +274,7 @@ Cross-cutting tests (run once):
Not Allowed (paivana rejects it, the
upstream is never contacted)
2 MiB POST upload request bodies above the 1 MiB
- REQUEST_BUFFER_MAX are rejected with
+ MAX_REQUEST_SIZE are rejected with
413 Content Too Large
curl keep-alive x3 three GETs over one keep-alive TCP
connection all succeed
@@ -269,8 +295,8 @@ Cross-cutting tests (run once):
with a 413 BEFORE reading the body.
Paivana must forward that 413 to the
client rather than turning it into a
- 502 — the early-response path in
- curl_download_cb / curl_upload_cb.
+ 502 — the early-response path, i.e.
+ #UP_DRAINING.
Note that the upstream is NOT expected
to see the whole body: RFC 9110 §9.3
lets a client stop sending once it has
@@ -290,13 +316,220 @@ Cross-cutting tests (run once):
the upstream's 413. Bounded by
timeout(1) rather than curl --max-time
because the failure mode is a hang:
- paivana's CURLOPT_TIMEOUT would
- eventually turn it into a 502 after
- 60s, which must not be allowed to look
- like a slow pass. The drain-mode case
- above cannot catch this — the upstream
- there keeps reading, so the socket
- never stays full.
+ paivana's stall watchdog would
+ eventually turn it into a truncated
+ response, which must not be allowed to
+ look like a slow pass. The drain-mode
+ case above cannot catch this — the
+ upstream there keeps reading, so the
+ socket never stays full.
+
+The streaming tests (`test_streaming`)
+--------------------------------------
+
+Everything above would pass equally well against the fully-buffered
+proxy this replaced: every body in it fits in one buffer. These
+cases are about what is new — that a body is no longer bounded by
+memory, and that it starts reaching the client before the origin has
+finished sending it.
+
+ 200 MiB, Content-Length five times the 40 MiB ceiling that used
+ to make this a 502 outright. Body
+ verified byte for byte, and the
+ origin's own Content-Length must reach
+ the client rather than one recomputed
+ from an assembled buffer.
+ 200 MiB, chunked the same body without a declared
+ length; must stay chunked to the
+ client instead of being silently
+ converted.
+ chunked to an HTTP/1.0 an HTTP/1.0 client cannot be sent
+ client chunks, so the close of the connection
+ has to be the framing.
+ Range -> 206 Content-Range and the partial body pass
+ through the streamed path.
+ HEAD on a large resource MHD does not run the content reader for
+ a HEAD but does emit the size the
+ response was created with, so the
+ length the equivalent GET would have
+ had now reaches the client (RFC 9110
+ §9.3.2). Buffering could only ever
+ report 0 here.
+ 204 / 304 no body either way; the 304 still
+ carries the length of the body it does
+ not send (RFC 9110 §8.6).
+ 200 MiB upload, both the request body is streamed too, so
+ framings the origin sees it byte-exact and sees
+ the client's own framing reproduced --
+ a declared length stays declared,
+ chunked stays chunked.
+ small POST the overwhelmingly common case, which
+ now takes the same path as the large
+ one.
+ chunked upstream stops the status is long gone by the time the
+ mid-stream origin gives up, so the only remaining
+ way to say "incomplete" is to close
+ without the terminating chunk. curl 18
+ is the client noticing.
+ upstream goes quiet MHD will not time this out (a suspended
+ connection is off its timeout lists)
+ and CURLOPT_TIMEOUT is deliberately
+ unset, so paivana's own stall watchdog
+ is the only thing that can end it.
+ upstream never answers distinct from an upstream that is not
+ there, which is a 502: this is a 504,
+ and the time-to-headers clock is what
+ tells them apart.
+ 100 abandoned downloads the ownership handshake between MHD's
+ completion notifier and the content
+ reader's free callback runs on every
+ request now, so a mistake in it is a
+ use-after-free or a leak on all
+ traffic. RSS across a hundred
+ abandoned transfers is the cheap
+ detector; ASan is the thorough one.
+ abandoned upload a Content-Length was declared upstream
+ that can no longer be delivered; the
+ origin has to be told the request is
+ broken rather than left waiting.
+ early 413 during a only reachable because the request body
+ 200 MiB upload is streamed: with it buffered first the
+ origin could not have answered before
+ seeing all of it.
+ trailers, 1xx what patch 0034 established, re-checked
+ on the streamed path: neither may be
+ merged into a response that has already
+ been queued.
+
+The congestion tests (`test_congestion`)
+----------------------------------------
+
+`test_streaming` shows that a large body gets through intact. It does
+not show that it got through *without being held in memory*, and every
+case in it would pass against a version that quietly buffered the lot
+-- so on their own they leave the central claim of the change untested.
+These are the cases that test it.
+
+Three things are measured that the client cannot see for itself:
+
+ paivana's VmRSS while a body many times the buffer size is in
+ flight. This is the bound, stated directly.
+
+ How long the *origin* took to write its body, which `stream_upstream`
+ reports per connection on stderr ("served target=... bytes=N ms=M").
+ A proxy that buffers takes everything at line rate however slowly its
+ own client reads; one that relays can only take what the client has
+ made room for. From the client end the two are indistinguishable,
+ which is why the origin has to report its own timing.
+
+ paivana's CPU time across an interval when nothing is moving.
+ Busy-waiting is the classic failure of a suspend/resume design and is
+ otherwise invisible: the transfer still completes, correctly, with a
+ core pinned for its duration.
+
+Rate limits (`--read-rate`, `--upload-rate` on the client, `rate=` on
+the upstream) are what make any of this reproducible. On loopback with
+both ends going flat out, the kernel socket buffers absorb everything
+and no ring ever fills.
+
+The RSS bounds are skipped under `--enable-sanitizers`. ASan's
+quarantine -- the thing that lets it catch a use-after-free -- holds
+freed chunks rather than reusing them, so RSS there tracks total bytes
+moved instead of bytes held: the 64 MiB case grows ~58 MB instrumented
+against ~0.5 MB not, for identical code. The transfers still run and
+LSan still watches them; the pacing and CPU assertions are unaffected
+and are checked in both builds.
+
+These sizes are deliberately *not* divided by PAIVANA_TEST_SCALE. Each
+case is rate-limited, so its duration is set by the rate and not by the
+size, and the sanitised build is no slower for them. Scaling them
+would also break the pacing assertions: the kernel socket buffers hold
+a fixed couple of megabytes however small the body is, so at a
+twentieth of the size the origin legitimately finishes well ahead of
+the client and "was it throttled" stops having a stable answer.
+
+ 64 MiB through a slow peak RSS over baseline must stay within a
+ client few megabytes. Measured: +552 kB across
+ 20 samples, against a hard 502 for this
+ size before the change. Checked against a
+ build with the ring cap removed, which
+ grows 12544k -> 78312k for the same body:
+ the bound does detect buffering.
+ upstream pacing the same transfer from the other end: the
+ origin's own elapsed time must track the
+ client's rather than finishing in a
+ fiftieth of it. Measured: 3597 ms to
+ write 64 MiB to a client that read for
+ 4000 ms, where buffering would have taken
+ under 100 ms on loopback.
+ upload pacing the same assertion in the request
+ direction, against /sink.
+ 32 concurrent throttled the per-request cost is what multiplies,
+ downloads so this is where a bound that holds for
+ one request and not for thirty-two shows.
+ Mixed rates, so the fast ones finish while
+ the slow ones are still going. Measured:
+ 10.6 MB of growth for 32 x 8 MiB in
+ flight, about 339 kB each. Runs with
+ PER_IP_CONNECTION_LIMIT lifted, which is
+ otherwise exactly 32 and would have the
+ case measure connection limiting instead.
+ idle transfer an origin dribbling 200 B/s leaves paivana
+ with nothing to do for ~5 s. CPU must
+ stay near zero (measured: 1 jiffy, i.e.
+ 10 ms, over 5107 ms), and the first byte
+ must still arrive at once -- measured at
+ 1 ms -- rather than after the last.
+ 1 KiB receive buffer makes libcurl drain paivana's socket in
+ tiny units, so MHD's content reader is
+ called hundreds of times where the default
+ buffer needs a handful -- each one a
+ chance for the ring to empty and the
+ connection to suspend and resume. Chunked,
+ so MHD's chunk framing is re-entered every
+ time.
+ slow at both ends neither side able to keep up with the
+ other on one request. Both rings spend
+ the transfer alternately full and empty
+ and the two halves of the state machine
+ have to interleave without deadlocking or
+ dropping a byte.
+
+The base64url cross-check (`test_base64url.sh`)
+-----------------------------------------------
+
+The paivana ID is `<expiration>-<base64url(sha256(...))>`. The daemon
+builds it with `GNUNET_STRINGS_base64url_encode()`; the browser rebuilds
+it in `paywall.js` to recognise the payment it has just made. Nothing
+in either program forces the two encoders to agree, and if they do not,
+the ID never matches, the payment appears not to go through, and
+neither side logs anything wrong.
+
+They have already disagreed twice. Once on the decode side: the daemon
+emits the RFC 4648 section 5 (URL-safe) alphabet, and the browser fed it
+to `atob()`, which only knows section 4 and throws on `-` or `_` -- at
+module scope, so the whole script died and the paywall could not be
+paid. Once on the encode side: the browser used
+`Uint8Array.prototype.toBase64`, which is a 2024-25 addition (Firefox
+133, Safari 18.2, Chrome 140) and is simply not a function on anything
+older. Two bugs of the same shape in one file is what this test is for.
+
+`base64url_vectors` prints 369 vectors as the *daemon* produces them --
+every length from 0 to 96, so both amounts of padding and none are
+crossed; every single byte value, because `-` and `_` are only
+reachable from particular high bit patterns and are exactly the two
+characters the section 4 alphabet spells differently; and sixteen
+32-byte blocks, that being the size which actually occurs.
+`test_base64url.sh` lifts `base64url()` out of `paywall.js` by matching
+braces -- rather than keeping a copy here, which would be a second
+implementation to hold in step, and holding implementations in step by
+hand is the thing that failed -- and compares. It refuses to pass on
+fewer than 300 vectors, so a generator that broke would fail rather
+than trivially agree. Skips (77) without node.
+
+Checked by breaking it: with the alphabet translation removed from
+`paywall.js`, it reports `paywall.js gave "WH2ix+wRNls", the daemon
+gives "WH2ix-wRNls"`.
The client_address unit test
----------------------------
@@ -446,7 +679,7 @@ Ports used
Every port is an offset off PAIVANA_PORT_BASE, which defaults to 18400
-- the 184xx / 185xx range, chosen to avoid collisions with real
-services. The suite checks all nine before it starts anything and
+services. The suite checks all ten before it starts anything and
exits 77 (meson reads that as SKIP) if one of them is taken, naming
it.
@@ -466,6 +699,7 @@ from exactly that probe.
base + 5 (18405) early_response_upstream
base + 6 (18406) early_response_upstream --no-drain
base + 7 (18407) truncating upstream (short-body test)
+ base + 8 (18408) stream_upstream (streaming tests)
base + 99 (18499) dead port (for "upstream down" test)
base + 100 (18500) paivana-httpd
diff --git a/src/tests/base64url_vectors.c b/src/tests/base64url_vectors.c
@@ -0,0 +1,109 @@
+/*
+ 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 base64url_vectors.c
+ * @brief Print base64url test vectors as the daemon produces them, for
+ * `test_base64url.sh' to check the paywall's JavaScript against.
+ *
+ * The paivana ID is `<expiration>-<base64url(sha256(...))>', and
+ * the browser recomputes it from the same inputs to recognise
+ * the payment it just made. Nothing forces the two encoders to
+ * agree, and they have already disagreed twice: once on the
+ * decode side, where the daemon emitted the RFC 4648 section 5
+ * alphabet and the browser fed it to atob(), which only knows
+ * section 4; and once on the encode side, where the browser
+ * used a method older engines do not have. Both broke the
+ * payment path and neither was visible from either side alone.
+ *
+ * Output is one vector per line, `<hex-input> <expected>', with
+ * an empty input written as `-'.
+ */
+#include "platform.h"
+#include <gnunet/gnunet_util_lib.h>
+
+
+/**
+ * Print one vector for the @a len bytes at @a buf.
+ */
+static void
+emit (const unsigned char *buf,
+ size_t len)
+{
+ char *out = NULL;
+ size_t n;
+
+ if (0 == len)
+ printf ("-");
+ for (size_t i = 0; i < len; i++)
+ printf ("%02x", buf[i]);
+ n = GNUNET_STRINGS_base64url_encode (buf,
+ len,
+ &out);
+ printf (" %.*s\n",
+ (int) n,
+ out);
+ GNUNET_free (out);
+}
+
+
+int
+main (int argc,
+ char **argv)
+{
+ unsigned char buf[96];
+
+ (void) argc;
+ (void) argv;
+ /* Every length from 0 to 96, which crosses the three-byte group
+ boundary in all its phases and so exercises both amounts of
+ padding as well as none. */
+ for (size_t len = 0; len <= sizeof (buf); len++)
+ {
+ for (size_t i = 0; i < len; i++)
+ buf[i] = (unsigned char) ((i * 37 + len * 11) & 0xFF);
+ emit (buf,
+ len);
+ }
+ /* Every single byte value, so no corner of the alphabet is missed --
+ '-' and '_' are only reachable from particular high bit patterns,
+ and they are exactly the two characters the section 4 alphabet
+ spells differently. */
+ for (unsigned int v = 0; v < 256; v++)
+ {
+ unsigned char b = (unsigned char) v;
+
+ emit (&b,
+ 1);
+ }
+ /* And a 32-byte block, which is the size that actually occurs: the
+ ID is a SHA-256 digest. */
+ for (unsigned int round = 0; round < 16; round++)
+ {
+ for (size_t i = 0; i < 32; i++)
+ buf[i] = (unsigned char) ((i * 101 + round * 251) & 0xFF);
+ emit (buf,
+ 32);
+ }
+ return 0;
+}
+
+
+/* end of base64url_vectors.c */
diff --git a/src/tests/lsan.supp b/src/tests/lsan.supp
@@ -0,0 +1,31 @@
+# LeakSanitizer suppressions for `./configure --enable-sanitizers'.
+#
+# Everything here is a one-time allocation in a library paivana links
+# against, made once per process and never freed because the library
+# has no teardown for it. None of it grows with the number of requests
+# served, which is what the sanitised run exists to catch.
+#
+# Keep this list short and specific. A suppression that names a
+# paivana symbol, or that is broad enough to cover one, defeats the
+# point of the mode: the ownership handshake between MHD's completion
+# notifier and the streaming content reader runs on every request, so a
+# mistake in it leaks on all traffic and must not be silenced here.
+
+# libgcrypt allocates its secure-memory pool and its RNG state at
+# gcry_check_version() and keeps them for the life of the process.
+leak:libgcrypt
+leak:gcry_
+
+# libcurl's global init (curl_global_init) and the OpenSSL/GnuTLS error
+# tables it pulls in are per-process and never released.
+leak:curl_global_init
+leak:libcurl
+
+# GnuTLS and its dependencies register global structures on first use.
+leak:libgnutls
+leak:gnutls_global_init
+
+# GNUnet's scheduler and configuration keep process-lifetime state; the
+# project-data descriptor in particular is deliberately immortal.
+leak:GNUNET_OS_init
+leak:GNUNET_SCHEDULER_run
diff --git a/src/tests/meson.build b/src/tests/meson.build
@@ -27,6 +27,27 @@ early_response_upstream = executable(
install: false,
)
+# The streaming pair: an origin that serves bodies larger than memory,
+# at a controlled rate and in framings a conforming server will not
+# emit, and a client that verifies them as they arrive rather than
+# storing them. Both generate the body from its offset, so a 200 MiB
+# case costs no disk on either side.
+stream_upstream = executable(
+ 'stream_upstream',
+ 'stream_upstream.c',
+ dependencies: [gnunetutil_dep],
+ include_directories: [incdir, configuration_inc],
+ install: false,
+)
+
+stream_client = executable(
+ 'stream_client',
+ 'stream_client.c',
+ dependencies: [curl_dep],
+ include_directories: [incdir, configuration_inc],
+ install: false,
+)
+
# Unit test for the client-address canonicalisation that the access
# cookie is keyed on. Links the two backend compilation units it
# exercises; the daemon's globals are supplied by the test itself.
@@ -57,10 +78,7 @@ test('client_address', test_client_address)
# unit; the daemon's globals are supplied by the test itself.
test_cookie_header = executable(
'test_cookie_header',
- [
- 'test_cookie_header.c',
- '../backend/paivana-httpd_cookie.c',
- ],
+ ['test_cookie_header.c', '../backend/paivana-httpd_cookie.c'],
dependencies: [
talerutil_dep,
talermhd_dep,
@@ -82,10 +100,7 @@ test('cookie_header', test_cookie_header)
# unit, with the daemon's globals supplied by the test.
test_cookie_access = executable(
'test_cookie_access',
- [
- 'test_cookie_access.c',
- '../backend/paivana-httpd_cookie.c',
- ],
+ ['test_cookie_access.c', '../backend/paivana-httpd_cookie.c'],
dependencies: [
talerutil_dep,
talermhd_dep,
@@ -101,10 +116,35 @@ test_cookie_access = executable(
test('cookie_access', test_cookie_access)
+# Prints base64url vectors as the daemon produces them, for
+# test_base64url.sh to check paywall.js against. The paivana ID
+# depends on the two encoders agreeing and nothing else forces them to;
+# they have already disagreed twice, both times breaking the payment
+# path.
+base64url_vectors = executable(
+ 'base64url_vectors',
+ 'base64url_vectors.c',
+ dependencies: [gnunetutil_dep],
+ include_directories: [incdir, configuration_inc],
+ install: false,
+)
+
+test(
+ 'base64url',
+ files('test_base64url.sh'),
+ env: {
+ 'SRCDIR': meson.current_source_dir(),
+ 'BUILDDIR': meson.current_build_dir(),
+ },
+ depends: [base64url_vectors],
+)
+
test_deps = [
upstream_mhd,
pipeline_client,
early_response_upstream,
+ stream_upstream,
+ stream_client,
paivana_httpd_exe,
]
@@ -132,16 +172,31 @@ if rustc_bin.found()
test_deps += upstream_rs
endif
+# Under `./configure --enable-sanitizers' the whole project is built
+# instrumented, so these are the same tests with ASan and UBSan under
+# them rather than extra cases. Two things are needed to keep that
+# mode usable rather than something everyone turns off again:
+#
+# - LSan needs the suppression file, or every run fails on one-time
+# allocations in libcurl and libgcrypt that are nothing to do with
+# paivana;
+# - the large cases have to scale down, since 200 MiB through ASan is
+# not a test but a coffee break. What those cases exercise is the
+# pause/resume interleaving, which is a function of the ring size
+# and not of the total, so a twentieth of the bytes reaches the same
+# transitions.
+test_env = {'SRCDIR': meson.current_source_dir(), 'BUILDDIR': meson.current_build_dir(), 'PAIVANA_HTTPD': paivana_httpd_exe.full_path(), 'LSAN_OPTIONS': 'suppressions='
+ + meson.current_source_dir() / 'lsan.supp'}
+if get_option('b_sanitize') != 'none'
+ test_env += {'PAIVANA_TEST_SCALE': '20', 'PAIVANA_SANITIZED': '1'}
+endif
+
test(
'reverse_proxy',
files('test_reverse_proxy.sh'),
- env: {
- 'SRCDIR': meson.current_source_dir(),
- 'BUILDDIR': meson.current_build_dir(),
- 'PAIVANA_HTTPD': paivana_httpd_exe.full_path(),
- },
+ env: test_env,
depends: test_deps,
- timeout: 300,
+ timeout: 600,
)
# The paywall against a real merchant backend. Skips (77) unless the
@@ -152,11 +207,7 @@ test(
test(
'paywall',
files('test_paywall.sh'),
- env: {
- 'SRCDIR': meson.current_source_dir(),
- 'BUILDDIR': meson.current_build_dir(),
- 'PAIVANA_HTTPD': paivana_httpd_exe.full_path(),
- },
+ env: test_env,
depends: test_deps,
timeout: 900,
)
diff --git a/src/tests/stream_client.c b/src/tests/stream_client.c
@@ -0,0 +1,463 @@
+/*
+ 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_client.c
+ * @brief HTTP client for the streaming tests: verifies a body against
+ * the generated pattern as it arrives, and reports what it saw
+ * about the *framing* rather than only about the bytes.
+ *
+ * Incremental verification is the point. A 200 MiB body
+ * written to a file and compared afterwards costs the disk and
+ * the wall clock of two extra copies, and — worse for these
+ * tests — says nothing about whether paivana streamed it or
+ * assembled it first. Here nothing larger than one libcurl
+ * delivery is ever held, so the client cannot be the thing that
+ * blows up the memory bound the tests are checking.
+ *
+ * What it prints on stdout, one `key=value' per line:
+ *
+ * status=N HTTP status
+ * bytes=N body bytes received
+ * pattern=ok|CORRUPT|... incremental pattern check
+ * chunked=yes|no Transfer-Encoding: chunked on the wire
+ * content_length=N|none the client-visible Content-Length
+ * curl=N CURLcode (0 = success, 18 = partial)
+ * ttfb_ms=N time to first body byte
+ * total_ms=N time to last byte
+ * body=... the response body, with --print-body
+ *
+ * The rate options are what make the congestion cases
+ * reproducible. Without a slow peer at one end, loopback and
+ * the kernel socket buffers absorb everything and no
+ * backpressure is ever exercised: paivana would look
+ * well-behaved whether or not it had any.
+ *
+ * The exit status is 0 whenever the exchange was carried out,
+ * *including* when the transfer failed: which failure occurred
+ * is data the driver asserts on, not a reason to abort. Exit 1
+ * is reserved for the client itself being unable to run.
+ *
+ * Usage:
+ * stream_client URL [--upload N] [--chunked-upload]
+ * [--expect-bytes N] [--head] [--print-body]
+ * [--abort-after N] [--read-rate N]
+ * [--upload-rate N] [--recv-buffer N]
+ */
+#include "platform.h"
+#include <curl/curl.h>
+#include <inttypes.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <strings.h>
+
+
+/**
+ * The body byte at offset @a off. Must agree with the same function
+ * in `stream_upstream.c'.
+ */
+static uint8_t
+pattern_at (uint64_t off)
+{
+ return (uint8_t) ((off * 7) + (off >> 13));
+}
+
+
+/**
+ * Body bytes received so far.
+ */
+static uint64_t received;
+
+/**
+ * Offset of the first byte that did not match the pattern, or
+ * UINT64_MAX if they all have.
+ */
+static uint64_t bad_at = UINT64_MAX;
+
+/**
+ * Did the response carry `Transfer-Encoding: chunked'?
+ */
+static bool saw_chunked;
+
+/**
+ * `Content-Length' of the response, or UINT64_MAX if there was none.
+ */
+static uint64_t saw_length = UINT64_MAX;
+
+/**
+ * Bytes of request body still to send.
+ */
+static uint64_t upload_left;
+
+/**
+ * Offset of the next request-body byte to send.
+ */
+static uint64_t upload_off;
+
+/**
+ * CURLOPT_BUFFERSIZE, or 0 to leave libcurl's default. A small value
+ * makes libcurl read the socket in small units, so paivana's content
+ * reader is called far more often and for far less each time --
+ * hundreds of pause/resume transitions where the default buffer would
+ * produce a handful.
+ */
+static unsigned long recv_buffer;
+
+/**
+ * Read the response body at no more than this many bytes per second,
+ * or 0 for as fast as it arrives.
+ */
+static uint64_t read_rate;
+
+/**
+ * Send the request body at no more than this many bytes per second,
+ * or 0 for as fast as libcurl will take it.
+ */
+static uint64_t upload_rate;
+
+/**
+ * Abort the transfer once this many body bytes have arrived, or 0
+ * never to. For the cases about what paivana does when its client
+ * walks away mid-response.
+ */
+static uint64_t abort_after;
+
+/**
+ * Print the response body rather than checking it against the
+ * pattern? For the targets whose answer is a short report about the
+ * request we sent, which is the interesting half of an upload case.
+ */
+static bool print_body;
+
+/**
+ * First #BODY_KEEP bytes of the response, when @e print_body is set.
+ */
+static char body_buf[512];
+
+/**
+ * Bytes held in @e body_buf.
+ */
+static size_t body_len;
+
+/**
+ * Time of the first body byte, or 0 if none has arrived.
+ */
+static struct timespec first_byte;
+
+/**
+ * When the request started.
+ */
+static struct timespec started;
+
+
+static long
+elapsed_ms (const struct timespec *from,
+ const struct timespec *to)
+{
+ return (long) ((to->tv_sec - from->tv_sec) * 1000
+ + (to->tv_nsec - from->tv_nsec) / 1000000);
+}
+
+
+/**
+ * Sleep for @a ms milliseconds, resuming across signals.
+ */
+static void
+sleep_ms (long ms)
+{
+ struct timespec ts;
+
+ if (0 >= ms)
+ return;
+ ts.tv_sec = ms / 1000;
+ ts.tv_nsec = (ms % 1000) * 1000000L;
+ while ( (0 != nanosleep (&ts, &ts)) &&
+ (EINTR == errno) )
+ ; /* again */
+}
+
+
+/**
+ * Hold @a moved bytes down to @a rate bytes per second, measuring from
+ * @a since.
+ *
+ * Paced against the clock rather than by a fixed nap per callback: the
+ * callback is handed whatever libcurl happened to read, so a fixed nap
+ * would make the actual rate depend on the delivery size and the cases
+ * that assert on elapsed time would be measuring the wrong thing.
+ *
+ * @param since when the transfer started
+ * @param moved bytes moved so far
+ * @param rate bytes per second to hold to
+ */
+static void
+pace (const struct timespec *since,
+ uint64_t moved,
+ uint64_t rate)
+{
+ struct timespec now;
+ long due_ms;
+ long spent_ms;
+
+ if (0 == rate)
+ return;
+ clock_gettime (CLOCK_MONOTONIC,
+ &now);
+ due_ms = (long) ((moved * 1000ULL) / rate);
+ spent_ms = elapsed_ms (since,
+ &now);
+ sleep_ms (due_ms - spent_ms);
+}
+
+
+static size_t
+write_cb (char *ptr,
+ size_t size,
+ size_t nmemb,
+ void *cls)
+{
+ size_t n = size * nmemb;
+
+ (void) cls;
+ if (0 == received)
+ clock_gettime (CLOCK_MONOTONIC,
+ &first_byte);
+ if (print_body)
+ {
+ size_t room = sizeof (body_buf) - body_len;
+ size_t take = (n < room) ? n : room;
+
+ memcpy (&body_buf[body_len],
+ ptr,
+ take);
+ body_len += take;
+ }
+ else
+ {
+ for (size_t i = 0; i < n; i++)
+ if ( (UINT64_MAX == bad_at) &&
+ ((uint8_t) ptr[i] != pattern_at (received + i)) )
+ bad_at = received + i;
+ }
+ received += n;
+ if ( (0 != abort_after) &&
+ (received >= abort_after) )
+ return 0; /* fail the transfer: this is us hanging up */
+ /* Blocking here is the whole point: libcurl stops reading the socket
+ while we sleep, its receive window closes, and paivana's own
+ download ring backs up. That is the congestion the cases are
+ about, and there is no way to produce it on loopback without it. */
+ pace (&first_byte,
+ received,
+ read_rate);
+ return n;
+}
+
+
+static size_t
+header_cb (char *ptr,
+ size_t size,
+ size_t nmemb,
+ void *cls)
+{
+ size_t n = size * nmemb;
+ static const char te[] = "transfer-encoding:";
+ static const char cl[] = "content-length:";
+
+ (void) cls;
+ if ( (n >= sizeof (te) - 1) &&
+ (0 == strncasecmp (ptr, te, sizeof (te) - 1)) )
+ {
+ /* The whole point of looking is to see the framing paivana chose,
+ so match the value rather than assume it. */
+ for (size_t i = sizeof (te) - 1; i + 6 < n; i++)
+ if (0 == strncasecmp (&ptr[i], "chunked", 7))
+ {
+ saw_chunked = true;
+ break;
+ }
+ }
+ if ( (n >= sizeof (cl) - 1) &&
+ (0 == strncasecmp (ptr, cl, sizeof (cl) - 1)) )
+ saw_length = strtoull (&ptr[sizeof (cl) - 1],
+ NULL,
+ 10);
+ return n;
+}
+
+
+static size_t
+read_cb (char *ptr,
+ size_t size,
+ size_t nmemb,
+ void *cls)
+{
+ size_t room = size * nmemb;
+ size_t n;
+
+ (void) cls;
+ n = (size_t) ((upload_left < (uint64_t) room)
+ ? upload_left
+ : (uint64_t) room);
+ for (size_t i = 0; i < n; i++)
+ ptr[i] = (char) pattern_at (upload_off + i);
+ upload_off += n;
+ upload_left -= n;
+ pace (&started,
+ upload_off,
+ upload_rate);
+ return n;
+}
+
+
+int
+main (int argc,
+ char **argv)
+{
+ CURL *eh;
+ CURLcode rc;
+ long status = 0;
+ const char *url = NULL;
+ uint64_t upload = 0;
+ bool chunked_upload = false;
+ bool head = false;
+ uint64_t expect = UINT64_MAX;
+ struct timespec done;
+ struct curl_slist *hdrs = NULL;
+
+ for (int i = 1; i < argc; i++)
+ {
+ if (0 == strcmp (argv[i], "--upload"))
+ upload = strtoull (argv[++i], NULL, 10);
+ else if (0 == strcmp (argv[i], "--chunked-upload"))
+ chunked_upload = true;
+ else if (0 == strcmp (argv[i], "--head"))
+ head = true;
+ else if (0 == strcmp (argv[i], "--print-body"))
+ print_body = true;
+ else if (0 == strcmp (argv[i], "--abort-after"))
+ abort_after = strtoull (argv[++i], NULL, 10);
+ else if (0 == strcmp (argv[i], "--read-rate"))
+ read_rate = strtoull (argv[++i], NULL, 10);
+ else if (0 == strcmp (argv[i], "--upload-rate"))
+ upload_rate = strtoull (argv[++i], NULL, 10);
+ else if (0 == strcmp (argv[i], "--recv-buffer"))
+ recv_buffer = strtoul (argv[++i], NULL, 10);
+ else if (0 == strcmp (argv[i], "--expect-bytes"))
+ expect = strtoull (argv[++i], NULL, 10);
+ else if (NULL == url)
+ url = argv[i];
+ else
+ {
+ fprintf (stderr,
+ "unexpected argument `%s'\n",
+ argv[i]);
+ return 1;
+ }
+ }
+ if (NULL == url)
+ {
+ fprintf (stderr,
+ "usage: %s URL [--upload N] [--chunked-upload]"
+ " [--head] [--print-body] [--expect-bytes N]"
+ " [--abort-after N] [--read-rate N] [--upload-rate N]"
+ " [--recv-buffer N]\n",
+ argv[0]);
+ return 1;
+ }
+ eh = curl_easy_init ();
+ if (NULL == eh)
+ {
+ fprintf (stderr,
+ "curl_easy_init failed\n");
+ return 1;
+ }
+ upload_left = upload;
+ curl_easy_setopt (eh, CURLOPT_URL, url);
+ curl_easy_setopt (eh, CURLOPT_WRITEFUNCTION, &write_cb);
+ curl_easy_setopt (eh, CURLOPT_HEADERFUNCTION, &header_cb);
+ curl_easy_setopt (eh, CURLOPT_NOSIGNAL, 1L);
+ if (0 != recv_buffer)
+ curl_easy_setopt (eh,
+ CURLOPT_BUFFERSIZE,
+ (long) recv_buffer);
+ if (head)
+ {
+ curl_easy_setopt (eh, CURLOPT_NOBODY, 1L);
+ }
+ else if (0 != upload)
+ {
+ curl_easy_setopt (eh, CURLOPT_POST, 1L);
+ curl_easy_setopt (eh, CURLOPT_READFUNCTION, &read_cb);
+ if (chunked_upload)
+ {
+ /* Say nothing about the length and libcurl chunks; this is the
+ framing paivana has to reproduce upstream. */
+ hdrs = curl_slist_append (hdrs,
+ "Transfer-Encoding: chunked");
+ curl_easy_setopt (eh, CURLOPT_HTTPHEADER, hdrs);
+ }
+ else
+ {
+ curl_easy_setopt (eh,
+ CURLOPT_POSTFIELDSIZE_LARGE,
+ (curl_off_t) upload);
+ }
+ }
+ clock_gettime (CLOCK_MONOTONIC, &started);
+ rc = curl_easy_perform (eh);
+ clock_gettime (CLOCK_MONOTONIC, &done);
+ curl_easy_getinfo (eh,
+ CURLINFO_RESPONSE_CODE,
+ &status);
+ printf ("status=%ld\n", status);
+ printf ("bytes=%" PRIu64 "\n", received);
+ if (print_body)
+ printf ("pattern=n/a\n");
+ else if (UINT64_MAX != bad_at)
+ printf ("pattern=CORRUPT at %" PRIu64 "\n", bad_at);
+ else if ( (UINT64_MAX != expect) &&
+ (received != expect) )
+ printf ("pattern=SHORT want %" PRIu64 "\n", expect);
+ else
+ printf ("pattern=ok\n");
+ printf ("chunked=%s\n", saw_chunked ? "yes" : "no");
+ if (UINT64_MAX == saw_length)
+ printf ("content_length=none\n");
+ else
+ printf ("content_length=%" PRIu64 "\n", saw_length);
+ printf ("curl=%d\n", (int) rc);
+ printf ("ttfb_ms=%ld\n",
+ (0 == received) ? -1 : elapsed_ms (&started, &first_byte));
+ printf ("total_ms=%ld\n", elapsed_ms (&started, &done));
+ if (print_body)
+ printf ("body=%.*s\n",
+ (int) body_len,
+ body_buf);
+ curl_slist_free_all (hdrs);
+ curl_easy_cleanup (eh);
+ return 0;
+}
+
+
+/* end of stream_client.c */
diff --git a/src/tests/stream_upstream.c b/src/tests/stream_upstream.c
@@ -0,0 +1,1263 @@
+/*
+ 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 */
diff --git a/src/tests/test_base64url.sh b/src/tests/test_base64url.sh
@@ -0,0 +1,104 @@
+#!/bin/sh
+# Check that the paywall's base64url encoder agrees with the daemon's.
+#
+# The paivana ID is "<expiration>-<base64url(sha256(...))>". The daemon
+# builds it with GNUNET_STRINGS_base64url_encode(); the browser rebuilds
+# it in paywall.js to recognise the payment it just made. If the two
+# spell base64url differently the ID never matches, the payment appears
+# not to go through, and neither side reports anything wrong -- so this
+# compares them directly, on vectors the daemon itself generates.
+#
+# It is worth having because the two have already diverged twice: once
+# on the decode side (the daemon emits the RFC 4648 section 5 alphabet,
+# atob() only understands section 4) and once on the encode side (the
+# browser used Uint8Array.prototype.toBase64, which older engines do not
+# have). Both broke the payment path.
+#
+# Exits 77 ("skipped") when there is no node to run the JavaScript in.
+#
+# Environment:
+# SRCDIR directory holding paywall.js (default: this script's ../frontend)
+# BUILDDIR directory holding base64url_vectors (default: $PWD)
+
+set -eu
+
+srcdir=${SRCDIR:-$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)}
+builddir=${BUILDDIR:-$PWD}
+frontend="$srcdir/../frontend"
+
+if ! command -v node >/dev/null 2>&1; then
+ echo "no node found, skipping the base64url cross-check" >&2
+ exit 77
+fi
+if [ ! -x "$builddir/base64url_vectors" ]; then
+ echo "base64url_vectors not built at $builddir" >&2
+ exit 1
+fi
+if [ ! -f "$frontend/paywall.js" ]; then
+ echo "paywall.js not found at $frontend" >&2
+ exit 1
+fi
+
+"$builddir/base64url_vectors" > "$builddir/base64url_vectors.txt"
+
+PAYWALL_JS="$frontend/paywall.js" \
+VECTORS="$builddir/base64url_vectors.txt" \
+node -e '
+const fs = require("fs");
+
+// Lift base64url() out of paywall.js rather than reimplementing it:
+// a copy here would be a second implementation to keep in step, and
+// this test exists precisely because keeping implementations in step
+// by hand is what failed. The module cannot simply be imported --
+// it reads window.location at load -- so the one function is cut out
+// by matching braces from its declaration.
+const src = fs.readFileSync(process.env.PAYWALL_JS, "utf8");
+const start = src.indexOf("function base64url(");
+if (start < 0) {
+ console.error("could not find base64url() in paywall.js;" +
+ " if it was renamed, update this test rather than delete it");
+ process.exit(1);
+}
+let depth = 0, end = -1, seen = false;
+for (let i = src.indexOf("{", start); i < src.length; i++) {
+ if (src[i] === "{") { depth++; seen = true; }
+ else if (src[i] === "}") { depth--; }
+ if (seen && depth === 0) { end = i + 1; break; }
+}
+if (end < 0) {
+ console.error("could not delimit base64url() in paywall.js");
+ process.exit(1);
+}
+const base64url = new Function(
+ `${src.slice(start, end)}; return base64url;`)();
+
+const lines = fs.readFileSync(process.env.VECTORS, "utf8")
+ .split("\n").filter((l) => l.length > 0);
+if (lines.length < 300) {
+ console.error(`only ${lines.length} vectors; expected the generator` +
+ " to produce hundreds -- refusing to pass vacuously");
+ process.exit(1);
+}
+let bad = 0;
+for (const line of lines) {
+ const sp = line.indexOf(" ");
+ const hex = line.slice(0, sp);
+ const want = line.slice(sp + 1);
+ const bytes = hex === "-"
+ ? new Uint8Array(0)
+ : Uint8Array.from(hex.match(/../g).map((h) => parseInt(h, 16)));
+ const got = base64url(bytes);
+ if (got !== want) {
+ if (bad < 5) {
+ console.error(`MISMATCH for ${hex}: paywall.js gave "${got}",` +
+ ` the daemon gives "${want}"`);
+ }
+ bad++;
+ }
+}
+if (bad > 0) {
+ console.error(`${bad} of ${lines.length} vectors disagree`);
+ process.exit(1);
+}
+console.log(`base64url agrees with the daemon on ${lines.length} vectors`);
+'
diff --git a/src/tests/test_reverse_proxy.sh b/src/tests/test_reverse_proxy.sh
@@ -91,6 +91,7 @@ RS_PORT=$((PORT_BASE + 4))
EARLY_PORT=$((PORT_BASE + 5))
NODRAIN_PORT=$((PORT_BASE + 6))
TRUNC_PORT=$((PORT_BASE + 7))
+STREAM_PORT=$((PORT_BASE + 8))
DEAD_PORT=$((PORT_BASE + 99)) # nothing may be listening here
PAIVANA_PORT=$((PORT_BASE + 100))
@@ -1042,13 +1043,701 @@ PYEOF
fi
}
+# ======================================================================
+# Streaming
+#
+# The point of the whole streaming path is that a proxied body is no
+# longer bounded by memory, and that it starts reaching the client
+# before the origin has finished sending. Neither is visible in the
+# battery above, where every body fits in one buffer and would have
+# done so before.
+#
+# `stream_upstream' serves bodies generated from their own offset, and
+# `stream_client' verifies them the same way as they arrive, so a
+# 200 MiB case costs no disk on either side and a duplicated or dropped
+# block is caught rather than just a wrong total. The client prints
+# `key=value' lines; `sfield' pulls one out.
+# ======================================================================
+
+# Bytes used for the "large" cases. 200 MiB is five times the 40 MiB
+# ceiling that used to make these a 502, which is the point.
+#
+# PAIVANA_TEST_SCALE divides it, for the sanitised build where 200 MiB
+# is a coffee break rather than a test. Dividing is sound here because
+# what these cases exercise is the pause/resume interleaving, and that
+# is a function of the ring size -- which does not scale -- rather than
+# of the total: a tenth of the bytes still crosses the ring hundreds of
+# times. It is not sound for the *bound* being checked, so the 413
+# cases and the buffer sizes are left alone.
+STREAM_BIG=$(( (200 * 1024 * 1024) / ${PAIVANA_TEST_SCALE:-1} ))
+
+function start_stream_upstream() {
+ local port="$1"
+ local log="$LOGDIR/stream.log"
+
+ ( exec "$BUILDDIR/stream_upstream" "$port" ) >"$log" 2>&1 &
+ local spid=$!
+ PIDS+=("$spid")
+ if ! wait_for_port 127.0.0.1 "$port" "$spid";
+ then
+ echo "FAIL: stream upstream did not start on port $port" >&2
+ tail -n 20 "$log" >&2
+ exit 1
+ fi
+}
+
+# Run stream_client against the streaming upstream through paivana and
+# leave its report in $SCRATCH/sc. Never fails the test itself: which
+# way a transfer went wrong is what the cases assert on.
+function sc() {
+ local path="$1"; shift
+
+ if ! timeout 300 "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "$path")" "$@" > "$SCRATCH/sc" 2>"$SCRATCH/scerr";
+ then
+ fail "stream_client did not finish for $path: $(cat "$SCRATCH/scerr")"
+ fi
+}
+
+# Value of one `key=value' line of the last sc() report.
+function sfield() {
+ sed -n "s/^$1=//p" "$SCRATCH/sc"
+}
+
+# Assert that a field of the last sc() report has the expected value.
+function sc_is() {
+ local key="$1" want="$2" got
+ got="$(sfield "$key")"
+ [ "$got" = "$want" ] || \
+ fail "$key=$got want=$want ($(tr '\n' ' ' < "$SCRATCH/sc"))"
+}
+
+function test_streaming() {
+ stop_paivana
+ start_stream_upstream "$STREAM_PORT"
+ # A short stall timeout so the "origin goes quiet" cases cost
+ # seconds rather than the minute the shipped default allows, and a
+ # request cap above the large upload cases.
+ local cfg="$SCRATCH/paivana-stream.conf"
+ sed -e "s|@DEST@|http://127.0.0.1:$STREAM_PORT|g" \
+ -e "s|@PORT@|$PAIVANA_PORT|g" \
+ "$SRCDIR/test_reverse_proxy.conf.in" > "$cfg"
+ cat >> "$cfg" <<EOF
+UPSTREAM_TIMEOUT = 3 s
+UPSTREAM_STALL_TIMEOUT = 3 s
+MAX_REQUEST_SIZE = $((512 * 1024 * 1024))
+EOF
+ PAIVANA_DEST="http://127.0.0.1:$STREAM_PORT"
+ local log="$LOGDIR/paivana.log"
+ ( exec "$PAIVANA_HTTPD" -c "$cfg" -n -L WARNING ) >"$log" 2>&1 &
+ PAIVANA_PID=$!
+ if ! wait_for_port 127.0.0.1 "$PAIVANA_PORT" "$PAIVANA_PID";
+ then
+ echo "FAIL: paivana-httpd did not start on port $PAIVANA_PORT" >&2
+ tail -n 20 "$log" >&2
+ exit 1
+ fi
+
+ # --- 1: a body five times the old ceiling, with a length --------
+ msg "streams a 200 MiB response with Content-Length"
+ sc "/cl?bytes=$STREAM_BIG" --expect-bytes "$STREAM_BIG"
+ sc_is status 200
+ sc_is curl 0
+ sc_is pattern ok
+ sc_is bytes "$STREAM_BIG"
+ # The origin's own framing, not one recomputed from a buffer.
+ sc_is content_length "$STREAM_BIG"
+ sc_is chunked no
+ ok
+
+ # --- 2: the same body, chunked ----------------------------------
+ msg "streams a 200 MiB chunked response, still chunked"
+ sc "/chunked?bytes=$STREAM_BIG" --expect-bytes "$STREAM_BIG"
+ sc_is status 200
+ sc_is curl 0
+ sc_is pattern ok
+ sc_is bytes "$STREAM_BIG"
+ # A chunked origin must not be silently converted to a declared
+ # length, which is what buffering the body did.
+ sc_is chunked yes
+ sc_is content_length none
+ ok
+
+ # --- 3: chunked origin, HTTP/1.0 client -------------------------
+ # An HTTP/1.0 client cannot be sent chunks, so the end of the body
+ # has to be the close of the connection.
+ msg "chunked upstream is close-delimited for an HTTP/1.0 client"
+ local out
+ out="$(curl -sS --http1.0 -o "$SCRATCH/h10" -D "$SCRATCH/h10hdr" \
+ -w '%{http_code}' --max-time 120 \
+ "$(PAIVANA_URL "/chunked?bytes=1048576")" 2>"$SCRATCH/err")" \
+ || fail "curl: $(cat "$SCRATCH/err")"
+ [ "$out" = "200" ] || fail "status=$out want=200"
+ grep -qi '^Transfer-Encoding:' "$SCRATCH/h10hdr" && \
+ fail "chunked encoding offered to an HTTP/1.0 client"
+ [ "$(wc -c < "$SCRATCH/h10")" = "1048576" ] || \
+ fail "got $(wc -c < "$SCRATCH/h10") bytes, want 1048576"
+ ok
+
+ # --- 4: a range request through the stream --------------------
+ msg "206 and Content-Range pass through a streamed response"
+ out="$(curl -sS -r 1000-1999 -o "$SCRATCH/rng" -D "$SCRATCH/rnghdr" \
+ -w '%{http_code}' --max-time 60 \
+ "$(PAIVANA_URL "/range?bytes=1048576")" 2>"$SCRATCH/err")" \
+ || fail "curl: $(cat "$SCRATCH/err")"
+ [ "$out" = "206" ] || fail "status=$out want=206"
+ grep -qi '^Content-Range: bytes 1000-1999/1048576' "$SCRATCH/rnghdr" || \
+ fail "no matching Content-Range: $(grep -i content-range "$SCRATCH/rnghdr")"
+ [ "$(wc -c < "$SCRATCH/rng")" = "1000" ] || \
+ fail "got $(wc -c < "$SCRATCH/rng") bytes, want 1000"
+ ok
+
+ # --- 5: HEAD on a large resource --------------------------------
+ # MHD does not run the content reader for a HEAD but does emit the
+ # size the response was created with, so the length the equivalent
+ # GET would have had now reaches the client (RFC 9110 9.3.2).
+ # Buffering could only ever have reported 0 here.
+ msg "HEAD reports the upstream's length without a body"
+ sc "/cl?bytes=$STREAM_BIG" --head
+ sc_is status 200
+ sc_is bytes 0
+ sc_is content_length "$STREAM_BIG"
+ ok
+
+ # --- 6: statuses that carry no body -----------------------------
+ msg "204 carries no body and no length"
+ sc "/status?code=204" --print-body
+ sc_is status 204
+ sc_is bytes 0
+ sc_is content_length none
+ ok
+
+ msg "304 keeps the length of the body it does not send"
+ sc "/status?code=304&len=12345" --print-body
+ sc_is status 304
+ sc_is bytes 0
+ sc_is content_length 12345
+ ok
+
+ # --- 7-8: the upload direction ----------------------------------
+ # The origin reports what it received; `pattern=ok' in its report
+ # is the byte-exactness assertion, and `framing=' is the assertion
+ # that the client's own framing was reproduced upstream rather
+ # than rewritten.
+ msg "streams a 200 MiB request body with Content-Length"
+ sc "/sink" --upload "$STREAM_BIG" --print-body
+ sc_is status 200
+ case "$(sfield body)" in
+ "bytes=$STREAM_BIG framing=length pattern=ok") ;;
+ *) fail "upstream saw: $(sfield body)" ;;
+ esac
+ ok
+
+ msg "streams a 200 MiB chunked request body, still chunked"
+ sc "/sink" --upload "$STREAM_BIG" --chunked-upload --print-body
+ sc_is status 200
+ case "$(sfield body)" in
+ "bytes=$STREAM_BIG framing=chunked pattern=ok") ;;
+ *) fail "upstream saw: $(sfield body)" ;;
+ esac
+ ok
+
+ # --- 9: the common case, which now takes the same path ----------
+ msg "a small POST still round-trips"
+ sc "/sink" --upload 100 --print-body
+ sc_is status 200
+ case "$(sfield body)" in
+ "bytes=100 framing=length pattern=ok") ;;
+ *) fail "upstream saw: $(sfield body)" ;;
+ esac
+ ok
+
+ # --- 18: chunked response with no terminating chunk -------------
+ # The status is long gone by the time the origin gives up, so the
+ # only remaining way to say "this is incomplete" is to close
+ # without the terminator. curl 18 is the client noticing.
+ msg "a chunked upstream that stops mid-stream truncates the client"
+ sc "/chunk-abort?after=5000"
+ sc_is status 200
+ sc_is chunked yes
+ sc_is bytes 5000
+ sc_is curl 18
+ ok
+
+ # --- 19: the stall watchdog -------------------------------------
+ # An origin that sends headers and some body and then goes quiet
+ # for ever. MHD will not time this out -- a suspended connection
+ # is off its timeout lists -- and CURLOPT_TIMEOUT is deliberately
+ # unset, so paivana's own watchdog is the only thing that can end
+ # it. Without it the client hangs until it gives up itself.
+ msg "an upstream that goes quiet is cut off by the stall watchdog"
+ local t0 t1
+ t0="$(date +%s)"
+ sc "/hang?after=1000"
+ t1="$(date +%s)"
+ sc_is status 200
+ sc_is bytes 1000
+ sc_is curl 18
+ [ "$((t1 - t0))" -lt 30 ] || \
+ fail "took $((t1 - t0))s; the 3 s stall timeout did not fire"
+ grep -q "moved no data" "$log" || \
+ fail "no stall diagnostic in the log"
+ ok
+
+ # --- 20: an upstream that accepts and never answers -------------
+ # Distinct from an upstream that is not there (502, tested
+ # separately): this one is a 504, and the time-to-headers clock is
+ # what tells them apart. It is the only one of the three clocks
+ # that can still produce a status code.
+ msg "an upstream that never answers yields 504"
+ out="$(curl -sS -o "$SCRATCH/body" -w '%{http_code}' --max-time 60 \
+ "$(PAIVANA_URL /mute)" 2>"$SCRATCH/err")" \
+ || fail "curl: $(cat "$SCRATCH/err")"
+ [ "$out" = "504" ] || fail "status=$out want=504"
+ ok
+
+ # --- 21: the client walks away mid-download ---------------------
+ # The interesting part is not the one request but that a hundred of
+ # them leave nothing behind: the response is queued and its content
+ # reader is live for every one of these, so a mistake in the
+ # ownership handshake between MHD's completion notifier and the
+ # reader's free callback leaks (or worse) on each.
+ msg "100 downloads abandoned mid-body leave no growth behind"
+ local rss0 rss1
+ sc "/cl?bytes=$STREAM_BIG" --abort-after 1048576
+ rss0="$(awk '/VmRSS/{print $2}' "/proc/$PAIVANA_PID/status")"
+ local i=0
+ while [ "$i" -lt 100 ];
+ do
+ timeout 60 "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "/cl?bytes=$STREAM_BIG")" \
+ --abort-after 1048576 >/dev/null 2>&1 || \
+ fail "stream_client did not finish on iteration $i"
+ i=$((i + 1))
+ done
+ rss1="$(awk '/VmRSS/{print $2}' "/proc/$PAIVANA_PID/status")"
+ if [ -n "${PAIVANA_SANITIZED:-}" ];
+ then
+ # RSS is not a leak detector under ASan: redzones around every
+ # allocation and a quarantine that deliberately withholds freed
+ # memory make the process grow whether or not anything leaked.
+ # The hundred iterations above still ran, and LSan is watching
+ # them -- which is a far better detector than this bound. It
+ # is this bound that is the stand-in, for the build where LSan
+ # is not there.
+ echo "OK (RSS bound not meaningful under sanitizers; LSan covers it)"
+ else
+ # A generous bound: the point is "flat", not "identical". Real
+ # per-request leakage of a ring or a response would be megabytes
+ # over a hundred iterations.
+ [ "$((rss1 - rss0))" -lt 4096 ] || \
+ fail "RSS grew ${rss0}k -> ${rss1}k over 100 abandoned downloads"
+ ok
+ fi
+
+ # --- 22: the client walks away mid-upload -----------------------
+ # We have declared a Content-Length upstream that we can no longer
+ # deliver, so the origin has to be told the request is broken
+ # rather than left waiting for bytes that will never come.
+ msg "an upload abandoned by the client does not wedge paivana"
+ timeout 60 curl -sS -o /dev/null --max-time 2 \
+ --data-binary "@$SCRATCH/echo_big" \
+ "$(PAIVANA_URL "/sink?rate=20000")" >/dev/null 2>&1
+ # Whatever that did to the one request, the daemon must still be
+ # serving; a wedged read callback would take the event loop with
+ # it.
+ sc "/cl?bytes=1024" --expect-bytes 1024
+ sc_is status 200
+ sc_is pattern ok
+ ok
+
+ # --- 23: the origin answers during a large upload ---------------
+ # Only reachable because the request body is streamed: with it
+ # buffered first, the origin could not have answered before seeing
+ # all of it. The client must get the origin's 413, not a 502, and
+ # the drain must complete rather than deadlock.
+ msg "an early 413 during a 200 MiB upload reaches the client"
+ sc "/sink-early?after=1048576" --upload "$STREAM_BIG" --print-body
+ sc_is status 413
+ ok
+
+ # --- 24-25: what 0034 established, on the streamed path ---------
+ msg "trailers on a streamed chunked response are still dropped"
+ out="$(curl -sS -o "$SCRATCH/tr" -D "$SCRATCH/trhdr" \
+ -w '%{http_code}' --max-time 60 \
+ "$(PAIVANA_URL "/trailers?bytes=4096")" 2>"$SCRATCH/err")" \
+ || fail "curl: $(cat "$SCRATCH/err")"
+ [ "$out" = "200" ] || fail "status=$out want=200"
+ grep -qi 'X-Trailer-Check' "$SCRATCH/trhdr" && \
+ fail "a trailer field was merged into the header section"
+ ok
+
+ msg "a 1xx before a streamed response is not merged into it"
+ out="$(curl -sS -o "$SCRATCH/ih" -D "$SCRATCH/ihhdr" \
+ -w '%{http_code}' --max-time 60 \
+ "$(PAIVANA_URL "/interim?bytes=4096")" 2>"$SCRATCH/err")" \
+ || fail "curl: $(cat "$SCRATCH/err")"
+ [ "$out" = "200" ] || fail "status=$out want=200"
+ grep -qi 'X-Interim-Check' "$SCRATCH/ihhdr" && \
+ fail "an interim-response header reappeared on the final response"
+ [ "$(wc -c < "$SCRATCH/ih")" = "4096" ] || \
+ fail "got $(wc -c < "$SCRATCH/ih") bytes, want 4096"
+ ok
+
+ stop_paivana
+}
+
+
+# ======================================================================
+# Congestion, and the bound being real
+#
+# The cases in test_streaming show that a large body gets through
+# intact. They do not show that it got through *without being held in
+# memory*, and they would all pass just as well against a version that
+# quietly buffered the lot -- so on their own the central claim of the
+# whole change is untested. These are the cases that test it.
+#
+# Three things are measured that the client cannot see on its own:
+#
+# - paivana's VmRSS while a large body is in flight, which is the
+# bound itself;
+# - how long the *origin* took to write its body, which is the
+# backpressure. A proxy that buffers takes everything at line rate
+# however slowly its client reads; one that relays can only take
+# what the client has made room for. From the client end the two
+# look identical, which is why the origin reports its own timing;
+# - paivana's CPU time across an interval when nothing is moving,
+# which is the busy-wait detector. Spinning is the classic failure
+# of a suspend/resume design and is otherwise invisible: the
+# transfer still completes, just with a core pinned.
+#
+# Rate limits are what make any of this reproducible. On loopback with
+# both ends going flat out, the kernel socket buffers absorb everything
+# and no ring ever fills.
+# ======================================================================
+
+# Sizes for these cases, in bytes. Smaller than test_streaming's,
+# because each is deliberately slowed to a few seconds and the point
+# here is the shape of the flow rather than the total.
+#
+# Deliberately NOT divided by PAIVANA_TEST_SCALE, unlike test_streaming.
+# Every one of these is rate-limited, so its duration is set by the rate
+# and not by the size, and the sanitised build is no slower for them.
+# Scaling them down would also break the pacing assertions outright: the
+# kernel socket buffers hold a fixed couple of megabytes however small
+# the body is, so at a twentieth of the size the origin legitimately
+# finishes well ahead of the client and "was it throttled" stops having
+# a stable answer.
+CONG_BIG=$((64 * 1024 * 1024))
+CONG_MID=$((32 * 1024 * 1024))
+CONG_SMALL=$((8 * 1024 * 1024))
+
+# Resident set of a process, in kB.
+function rss_kb() {
+ awk '/VmRSS/{print $2}' "/proc/$1/status" 2>/dev/null || echo 0
+}
+
+# User+system CPU of a process, in jiffies (100 per second).
+function cpu_jiffies() {
+ awk '{print $14 + $15}' "/proc/$1/stat" 2>/dev/null || echo 0
+}
+
+# Sample rss_kb of $1 every 200 ms until none of the pids in $2.. are
+# left, leaving the maximum in $PEAK_RSS and the number of samples
+# taken in $RSS_SAMPLES.
+#
+# Deliberately given the pids to wait for rather than asking `jobs':
+# paivana itself is a background job of this same shell, so "wait while
+# any job is running" never becomes false and the sampler spins for
+# ever. For the same reason callers must `wait' on the client pids by
+# name and not bare.
+#
+# The sample count is not bookkeeping. If the transfer finishes before
+# the first tick, the loop never runs, the peak is whatever RSS was
+# before it started, and "no growth" is asserted about a measurement
+# that was never taken -- a silent vacuous pass on the one claim the
+# whole change rests on. Callers must check it.
+function watch_rss() {
+ local pid="$1"; shift
+ local r p alive
+ PEAK_RSS="$(rss_kb "$pid")"
+ RSS_SAMPLES=0
+ while true;
+ do
+ alive=0
+ for p in "$@";
+ do
+ if kill -0 "$p" 2>/dev/null;
+ then
+ alive=1
+ break
+ fi
+ done
+ [ "$alive" = "0" ] && break
+ r="$(rss_kb "$pid")"
+ [ "${r:-0}" -gt "$PEAK_RSS" ] && PEAK_RSS="$r"
+ RSS_SAMPLES=$((RSS_SAMPLES + 1))
+ sleep 0.2
+ done
+}
+
+# Assert $PEAK_RSS is no more than $2 kB above the baseline $1, and
+# report the measurement; $3 describes the transfer for the failure.
+#
+# Skipped under sanitizers, where RSS stops meaning what this case
+# needs it to mean. ASan surrounds every allocation with redzones and,
+# more to the point, holds freed chunks in a quarantine rather than
+# reusing them -- that quarantine is exactly what lets it catch a
+# use-after-free, so turning it down to make this number readable would
+# trade away the thing the sanitised build exists for. Measured: the
+# 64 MiB case grows ~58 MB instrumented against ~0.5 MB not, for
+# identical code. LSan is the detector in that build; this is its
+# stand-in in the ordinary one.
+#
+# Printing the number rather than just "OK" is deliberate: a bound that
+# is never approached and a bound that was never measured look the same
+# from a pass, and this is the assertion the whole change rests on.
+function rss_bound() {
+ local base="$1" limit="$2" what="$3"
+
+ if [ -n "${PAIVANA_SANITIZED:-}" ];
+ then
+ echo "OK (RSS bound not meaningful under sanitizers;" \
+ "saw +$((PEAK_RSS - base))k, LSan covers the leak side)"
+ return
+ fi
+ [ "$((PEAK_RSS - base))" -lt "$limit" ] || \
+ fail "RSS grew ${base}k -> ${PEAK_RSS}k $what"
+ echo "OK (peak +$((PEAK_RSS - base))k over $RSS_SAMPLES samples)"
+}
+
+# ms the origin reported for the most recent request whose target
+# matches $1. See the "served" line stream_upstream writes per
+# connection.
+function origin_ms() {
+ tr -d '\0' < "$LOGDIR/stream.log" \
+ | grep -a "^served target=$1 " | tail -1 \
+ | sed -n 's/.* ms=//p'
+}
+
+function origin_bytes() {
+ tr -d '\0' < "$LOGDIR/stream.log" \
+ | grep -a "^served target=$1 " | tail -1 \
+ | sed -n 's/.* bytes=\([0-9]*\) .*/\1/p'
+}
+
+function test_congestion() {
+ stop_paivana
+ # Points at the upstream test_streaming already started. Its own
+ # paivana, so the buffer sizes these cases assert against are
+ # stated here rather than inherited.
+ local cfg="$SCRATCH/paivana-congestion.conf"
+ sed -e "s|@DEST@|http://127.0.0.1:$STREAM_PORT|g" \
+ -e "s|@PORT@|$PAIVANA_PORT|g" \
+ "$SRCDIR/test_reverse_proxy.conf.in" > "$cfg"
+ cat >> "$cfg" <<EOF
+REQUEST_BUFFER_MAX = 262144
+RESPONSE_BUFFER_MAX = 262144
+MAX_REQUEST_SIZE = $((512 * 1024 * 1024))
+UPSTREAM_STALL_TIMEOUT = 30 s
+# The concurrency case runs 32 transfers from 127.0.0.1, which is
+# exactly PER_IP_CONNECTION_LIMIT's default: leaving it would have the
+# case measure connection limiting rather than the memory bound, and
+# would do so by refusing whichever request happened to be 33rd.
+PER_IP_CONNECTION_LIMIT = 0
+EOF
+ PAIVANA_DEST="http://127.0.0.1:$STREAM_PORT"
+ local log="$LOGDIR/paivana.log"
+ ( exec "$PAIVANA_HTTPD" -c "$cfg" -n -L WARNING ) >"$log" 2>&1 &
+ PAIVANA_PID=$!
+ if ! wait_for_port 127.0.0.1 "$PAIVANA_PORT" "$PAIVANA_PID";
+ then
+ echo "FAIL: paivana-httpd did not start on port $PAIVANA_PORT" >&2
+ tail -n 20 "$log" >&2
+ exit 1
+ fi
+
+ # --- 10: the bound itself ---------------------------------------
+ # A body many times the size of the buffers, through a client slow
+ # enough that paivana cannot simply hand it straight on. Peak RSS
+ # over baseline is the assertion, and it is the one that makes the
+ # rest of the suite mean anything: everything else here would pass
+ # against a version that buffered the whole body.
+ msg "a $((CONG_BIG / 1024 / 1024)) MiB download through a slow client stays within its buffers"
+ local base peak
+ base="$(rss_kb "$PAIVANA_PID")"
+ "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "/cl?bytes=$CONG_BIG")" \
+ --expect-bytes "$CONG_BIG" \
+ --read-rate $((16 * 1024 * 1024)) > "$SCRATCH/sc" 2>&1 &
+ local cpid=$!
+ watch_rss "$PAIVANA_PID" "$cpid"
+ wait "$cpid" || fail "stream_client did not finish"
+ sc_is status 200
+ sc_is pattern ok
+ sc_is bytes "$CONG_BIG"
+ # Generous: two 256 KiB rings, MHD's block buffer, libcurl's own
+ # buffering and glibc's allocator. The number that matters is that
+ # it does not scale with the body -- buffering would show tens of
+ # megabytes here, and CONG_BIG is far above the ceiling that used to
+ # apply at all.
+ [ "$RSS_SAMPLES" -ge 3 ] || \
+ fail "only $RSS_SAMPLES RSS samples taken; the transfer was too fast to have measured anything"
+ rss_bound "$base" 16384 "relaying $CONG_BIG bytes"
+
+ # --- 11: the origin really was held back ------------------------
+ # Same transfer, seen from the other end. Without backpressure the
+ # origin writes its body at loopback speed -- well under a second
+ # for this size -- and paivana holds the difference. With it, the
+ # origin can only get as far ahead as the buffers allow, so its own
+ # elapsed time tracks the client's.
+ msg "the upstream is paced by the client rather than by the socket"
+ local oms cms
+ oms="$(origin_ms "/cl?bytes=$CONG_BIG")"
+ cms="$(sfield total_ms)"
+ [ -n "$oms" ] || fail "upstream reported no timing for /cl?bytes=$CONG_BIG"
+ [ "$(origin_bytes "/cl?bytes=$CONG_BIG")" = "$CONG_BIG" ] || \
+ fail "upstream wrote $(origin_bytes "/cl?bytes=$CONG_BIG") of $CONG_BIG bytes"
+ # Half is the slack for the kernel socket buffers on both sides plus
+ # the rings; the failure this is looking for is the origin finishing
+ # in a fiftieth of the time, not in nine tenths of it.
+ [ "$oms" -ge "$((cms / 2))" ] || \
+ fail "upstream finished writing in ${oms}ms while the client took ${cms}ms: it was not throttled"
+ echo "OK (upstream ${oms}ms, client ${cms}ms)"
+
+ # --- 12: the same, in the upload direction ----------------------
+ msg "a slow client uploading is likewise paced end to end"
+ "$BUILDDIR/stream_client" "$(PAIVANA_URL /sink)" \
+ --upload "$CONG_MID" --upload-rate $((16 * 1024 * 1024)) \
+ --print-body > "$SCRATCH/sc" 2>&1 \
+ || fail "stream_client did not finish"
+ sc_is status 200
+ case "$(sfield body)" in
+ "bytes=$CONG_MID framing=length pattern=ok") ;;
+ *) fail "upstream saw: $(sfield body)" ;;
+ esac
+ oms="$(origin_ms "/sink")"
+ cms="$(sfield total_ms)"
+ [ -n "$oms" ] || fail "upstream reported no timing for /sink"
+ [ "$oms" -ge "$((cms / 2))" ] || \
+ fail "upstream finished reading in ${oms}ms while the client took ${cms}ms"
+ echo "OK (upstream ${oms}ms, client ${cms}ms)"
+
+ # --- 13: many at once -------------------------------------------
+ # The per-request cost is what multiplies, so this is where a bound
+ # that holds for one request and not for sixteen would show. Mixed
+ # rates so the fast ones finish while the slow ones are still going,
+ # which is the state a single-rate run never reaches.
+ msg "32 concurrent throttled downloads stay within a bounded total"
+ base="$(rss_kb "$PAIVANA_PID")"
+ rm -f "$SCRATCH"/cong.*.out
+ local i
+ local cpids=""
+ for i in $(seq 1 32);
+ do
+ "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "/cl?bytes=$CONG_SMALL")" \
+ --expect-bytes "$CONG_SMALL" \
+ --read-rate $(( (i % 4 + 1) * 4 * 1024 * 1024 )) \
+ > "$SCRATCH/cong.$i.out" 2>&1 &
+ cpids="$cpids $!"
+ done
+ # shellcheck disable=SC2086
+ watch_rss "$PAIVANA_PID" $cpids
+ # shellcheck disable=SC2086
+ wait $cpids
+ local okcount
+ okcount="$(cat "$SCRATCH"/cong.*.out | grep -c '^pattern=ok$')"
+ [ "$okcount" = "32" ] || \
+ fail "only $okcount of 32 concurrent bodies verified"
+ grep -hq '^status=200$' "$SCRATCH"/cong.1.out || fail "no 200 seen"
+ # 32 requests times two 256 KiB rings is 16 MiB of ceiling; the
+ # buffered path would have held 32 times the body instead.
+ [ "$RSS_SAMPLES" -ge 3 ] || \
+ fail "only $RSS_SAMPLES RSS samples taken across the concurrent transfers"
+ rss_bound "$base" 32768 \
+ "across 32 concurrent transfers of $((CONG_SMALL / 1024 / 1024)) MiB"
+
+ # --- 14: not busy-waiting ---------------------------------------
+ # An origin dribbling a byte at a time means paivana spends almost
+ # the whole request with nothing to do. If suspend/resume is wrong
+ # -- an MHD content reader returning 0 without suspending, or an
+ # unpause that reschedules itself -- the transfer still completes
+ # and nothing else in the suite notices; a core is simply pinned for
+ # the duration. CPU time is the only thing that shows it.
+ msg "an idle transfer costs no CPU, and its first byte still arrives at once"
+ local c0 c1 jiffies
+ c0="$(cpu_jiffies "$PAIVANA_PID")"
+ "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "/cl?bytes=1024&rate=200")" \
+ --expect-bytes 1024 > "$SCRATCH/sc" 2>&1 \
+ || fail "stream_client did not finish"
+ c1="$(cpu_jiffies "$PAIVANA_PID")"
+ sc_is status 200
+ sc_is pattern ok
+ jiffies=$((c1 - c0))
+ # ~5 s of wall clock, i.e. ~500 jiffies were available to burn.
+ [ "$jiffies" -lt 100 ] || \
+ fail "paivana used ${jiffies} jiffies of CPU across a ~5 s idle transfer; it is spinning"
+ # And the point of streaming at all: the client does not wait for
+ # the origin's last byte to see its first.
+ [ "$(sfield ttfb_ms)" -lt 1000 ] || \
+ fail "first byte took $(sfield ttfb_ms)ms of a ~5 s transfer; the body was buffered"
+ echo "OK (${jiffies} jiffies CPU, first byte at $(sfield ttfb_ms)ms of $(sfield total_ms)ms)"
+
+ # --- 15: pathological interleaving ------------------------------
+ # A 1 KiB receive buffer makes libcurl drain paivana's socket in
+ # tiny units, so MHD's content reader is called hundreds of times
+ # for a body the default buffer would move in a handful -- and each
+ # of those is a chance for the ring to empty and the connection to
+ # suspend and resume. Chunked, because that path re-enters MHD's
+ # chunk framing on every one of them.
+ msg "a client reading in 1 KiB units survives the pause/resume churn"
+ "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "/chunked?bytes=$CONG_SMALL")" \
+ --expect-bytes "$CONG_SMALL" --recv-buffer 1024 \
+ > "$SCRATCH/sc" 2>&1 \
+ || fail "stream_client did not finish"
+ sc_is status 200
+ sc_is chunked yes
+ sc_is pattern ok
+ sc_is bytes "$CONG_SMALL"
+ ok
+
+ # --- 16: slow at both ends simultaneously -----------------------
+ # Neither side able to keep up with the other, on the same request.
+ # Both rings spend the transfer alternately full and empty, and the
+ # two halves of the state machine have to interleave without either
+ # deadlocking or dropping a byte.
+ msg "a slow upstream and a slow client on one request"
+ "$BUILDDIR/stream_client" \
+ "$(PAIVANA_URL "/cl?bytes=$CONG_SMALL&rate=$((8 * 1024 * 1024))")" \
+ --expect-bytes "$CONG_SMALL" --read-rate $((4 * 1024 * 1024)) \
+ > "$SCRATCH/sc" 2>&1 \
+ || fail "stream_client did not finish"
+ sc_is status 200
+ sc_is pattern ok
+ sc_is bytes "$CONG_SMALL"
+ ok
+
+ stop_paivana
+}
+
+
function test_short_body() {
# An incomplete message must be treated as a failure (RFC 9112
- # section 8.1.2). Without that, paivana served the fragment under
- # the upstream's 200 with a Content-Length MHD had computed for
- # what was left -- a corrupt representation the client has no way
- # of telling apart from a complete one.
- msg "upstream body shorter than its Content-Length yields 502"
+ # section 8.1.2), and with a streamed response the only way left to
+ # say so is to break the framing. The upstream's status and its
+ # Content-Length reach the client long before the body runs out --
+ # they cannot be retracted afterwards -- so the client sees a 200
+ # promising 100000 bytes and a connection that closes after 1000.
+ # curl reports that as error 18, which is the assertion here.
+ #
+ # A 502 used to be possible because the whole body was assembled
+ # before anything was sent. What must never happen either way is a
+ # short body served as if it were complete: that is what the
+ # "bytes remaining" check below rules out.
+ msg "upstream body shorter than its Content-Length truncates the client"
if ! command -v python3 >/dev/null 2>&1;
then
echo "SKIP (python3 missing)"
@@ -1057,15 +1746,22 @@ function test_short_body() {
stop_paivana
start_truncating_upstream "$TRUNC_PORT"
start_paivana "http://127.0.0.1:$TRUNC_PORT"
- local status
- status="$(curl -sS -o "$SCRATCH/body" -w '%{http_code}' \
- --max-time 30 \
- "$(PAIVANA_URL /short)" 2>"$SCRATCH/err")" \
- || fail "curl: $(cat "$SCRATCH/err")"
- [ "$status" = "502" ] || \
- fail "status=$status want=502 (a 200 here means the partial body was served as if complete)"
- grep -qi 'bad gateway' "$SCRATCH/body" || \
- fail "no 'Bad Gateway' in body"
+ local out status
+ # curl exits 18 (CURLE_PARTIAL_FILE) here, so the pipeline must not
+ # be allowed to fail the test; the exit code is part of what is
+ # being asserted.
+ out="$(curl -sS -o "$SCRATCH/body" -D "$SCRATCH/hdr" \
+ -w '%{http_code}' --max-time 30 \
+ "$(PAIVANA_URL /short)" 2>"$SCRATCH/err")"
+ status=$?
+ [ "$status" = "18" ] || \
+ fail "curl exit=$status want=18 (partial file); a 0 here means the truncation was hidden from the client"
+ [ "$out" = "200" ] || \
+ fail "status=$out want=200 (the upstream's status is already sent when the body runs out)"
+ grep -qi '^Content-Length: 100000' "$SCRATCH/hdr" || \
+ fail "client was not told the upstream's declared length: $(grep -i content-length "$SCRATCH/hdr")"
+ [ "$(wc -c < "$SCRATCH/body")" = "1000" ] || \
+ fail "got $(wc -c < "$SCRATCH/body") bytes, want the 1000 the upstream actually sent"
ok
stop_paivana
}
@@ -1572,11 +2268,11 @@ echo "Temp dir: $SCRATCH"
echo "Paivana binary: $PAIVANA_HTTPD"
echo "Source dir: $SRCDIR"
echo "Build dir: $BUILDDIR"
-echo "Ports: $PORT_BASE + 1..7, 99, 100"
+echo "Ports: $PORT_BASE + 1..8, 99, 100"
require_ports_free "$MHD_PORT" "$GO_PORT" "$PY_PORT" "$RS_PORT" \
"$EARLY_PORT" "$NODRAIN_PORT" "$TRUNC_PORT" \
- "$DEAD_PORT" "$PAIVANA_PORT"
+ "$STREAM_PORT" "$DEAD_PORT" "$PAIVANA_PORT"
start_upstreams
@@ -1634,6 +2330,12 @@ fi
test_early_response
test_early_response_no_drain
+# --- Streaming (restarts paivana) -------------------------------------
+test_streaming
+
+# --- Congestion and the memory bound (restarts paivana) ---------------
+test_congestion
+
# --- Truncated-response test (restarts paivana) -----------------------
test_short_body
