From 43e976775d0554b9889fffccfe2f20a4491188a6 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Tue, 13 May 2025 17:19:46 -0600
Subject: [PATCH 01/15] update some logging, add comments

---
 build.savant                                  |  2 +-
 java-http.ipr                                 | 15 ++++++++
 .../server/internal/HTTPServerThread.java     | 36 ++++++++++++-------
 .../http/server/internal/HTTPWorker.java      | 33 +++++++++--------
 .../fusionauth/http/server/io/Throughput.java |  9 +++--
 .../http/io/MultipartStreamTest.java          | 23 ++++++++++--
 6 files changed, 84 insertions(+), 34 deletions(-)

diff --git a/build.savant b/build.savant
index 6ff5a65..da1bc59 100644
--- a/build.savant
+++ b/build.savant
@@ -17,7 +17,7 @@ jackson5Version = "3.0.1"
 restifyVersion = "4.2.1"
 testngVersion = "7.10.2"
 
-project(group: "io.fusionauth", name: "java-http", version: "0.4.0-RC.4", licenses: ["ApacheV2_0"]) {
+project(group: "io.fusionauth", name: "java-http", version: "0.4.0-RC.5", licenses: ["ApacheV2_0"]) {
   workflow {
     fetch {
       // Dependency resolution order:
diff --git a/java-http.ipr b/java-http.ipr
index 3b37e61..d48a888 100644
--- a/java-http.ipr
+++ b/java-http.ipr
@@ -358,12 +358,14 @@
       </HTMLCodeStyleSettings>
       <JavaCodeStyleSettings>
         <option name="DO_NOT_WRAP_AFTER_SINGLE_ANNOTATION" value="true" />
+        <option name="BLANK_LINES_AROUND_FIELD_WITH_ANNOTATIONS" value="1" />
         <option name="CLASS_NAMES_IN_JAVADOC" value="3" />
         <option name="INSERT_INNER_CLASS_IMPORTS" value="true" />
         <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="9999" />
         <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="9999" />
         <option name="IMPORT_LAYOUT_TABLE">
           <value>
+            <package name="" withSubpackages="true" static="false" module="true" />
             <package name="javax" withSubpackages="true" static="false" />
             <package name="java" withSubpackages="true" static="false" />
             <emptyLine />
@@ -1409,4 +1411,17 @@
   <component name="VcsDirectoryMappings">
     <mapping directory="$PROJECT_DIR$" vcs="Git" />
   </component>
+  <component name="libraryTable">
+    <library name=".kts definition dependencies">
+      <CLASSES>
+        <root url="jar://$KOTLIN_BUNDLED$/lib/kotlin-reflect.jar!/" />
+        <root url="jar://$KOTLIN_BUNDLED$/lib/kotlin-script-runtime.jar!/" />
+        <root url="jar://$KOTLIN_BUNDLED$/lib/kotlin-stdlib.jar!/" />
+      </CLASSES>
+      <JAVADOC />
+      <SOURCES>
+        <root url="jar://$KOTLIN_BUNDLED$/lib/kotlin-stdlib-sources.jar!/" />
+      </SOURCES>
+    </library>
+  </component>
 </project>
\ No newline at end of file
diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
index 8696d20..df8d4b4 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -157,14 +157,15 @@ public HTTPServerCleanerThread() {
 
     public void run() {
       while (running) {
-        logger.debug("Cleaning things up");
+        logger.trace("Wake up. Clean up server threads.");
 
         Iterator<ClientInfo> iterator = clients.iterator();
         while (iterator.hasNext()) {
           ClientInfo client = iterator.next();
           Thread thread = client.thread();
+          long threadId = thread.threadId();
           if (!thread.isAlive()) {
-            logger.debug("Thread is dead. Removing.");
+            logger.debug("[{}] Thread is dead. Removing.", threadId);
             iterator.remove();
             continue;
           }
@@ -179,22 +180,33 @@ public void run() {
           boolean timedOut = false;
           boolean badClient = false;
 
+          // -1 means we have not yet calculated these values
+          long readThroughput = -1;
+          long writeThroughput = -1;
+
+          String badClientReason = "[" + threadId + "] Checking worker in state [" + state + "]";
           if (state == HTTPWorker.State.Read) {
             // Here the SO_TIMEOUT set above or the Keep-Alive timeout in HTTPWorker will dictate if the socket has timed out. This prevents slow readers
             // or network issues where the client reads 1 byte per timeout value (i.e. 1 byte per 2 seconds or something like that)
-            badClient = throughput.readThroughput(now) < minimumReadThroughput;
+            readThroughput = throughput.readThroughput(now);
+            badClient = readThroughput < minimumReadThroughput;
             readingSlow = badClient;
+            badClientReason += " readingSlow=[" + readingSlow + "] readThroughput=[" + readThroughput + "] minimumReadThroughput=[" + minimumReadThroughput + "]";
           } else if (state == HTTPWorker.State.Write) {
             // Check for slow clients when writing (or network issues)
-            badClient = throughput.writeThroughput(now) < minimumWriteThroughput;
+            writeThroughput = throughput.writeThroughput(now);
+            badClient = writeThroughput < minimumWriteThroughput;
             writingSlow = badClient;
+            badClientReason += " writingSlow=[" + writingSlow + "] writeThroughput=[" + writeThroughput + "] minimumWriteThroughput=[" + minimumWriteThroughput + "]";
           } else if (state == HTTPWorker.State.Process) {
             // Here lastUsed was the instant the last byte was read, so we calculate distance between that and now to see if it is beyond the timeout
-            badClient = now - workerLastUsed > configuration.getProcessingTimeoutDuration().toMillis();
+            long waited = (now - workerLastUsed);
+            badClient = waited > configuration.getProcessingTimeoutDuration().toMillis();
             timedOut = badClient;
+            badClientReason += " timedOut=[" + timedOut + "] waited=[" + waited + "] processingTimeoutDuration=[" + configuration.getProcessingTimeoutDuration().toMillis() + "]";
           }
 
-          logger.debug("Checking client readingSlow=[{}] writingSlow=[{}] timedOut=[{}] writeThroughput=[{}] minWriteThroughput=[{}]", readingSlow, writingSlow, timedOut, throughput.writeThroughput(now), minimumWriteThroughput);
+          logger.debug(badClientReason);
 
           if (!badClient) {
             continue;
@@ -207,19 +219,19 @@ public void run() {
             String message = "";
 
             if (readingSlow) {
-              message += String.format(" Min read throughput [%s], actual throughput [%s].", minimumReadThroughput, throughput.readThroughput(now));
+              message += String.format(" Min read throughput [%s], actual throughput [%s].", minimumReadThroughput, readThroughput);
             }
 
             if (writingSlow) {
-              message += String.format(" Min write throughput [%s], actual throughput [%s].", minimumWriteThroughput, throughput.writeThroughput(now));
+              message += String.format(" Min write throughput [%s], actual throughput [%s].", minimumWriteThroughput, writeThroughput);
             }
 
             if (timedOut) {
               message += String.format(" Connection timed out while processing. Last used [%s]ms ago. Configured client timeout [%s]ms.", (now - workerLastUsed), configuration.getProcessingTimeoutDuration().toMillis());
             }
 
-            logger.debug("Closing connection readingSlow=[{}] writingSlow=[{}] timedOut=[{}] {}", readingSlow, writingSlow, timedOut, message);
-            logger.debug("Closing client connection [{}] due to inactivity", worker.getSocket().getRemoteSocketAddress());
+            logger.debug("[{}] Closing connection readingSlow=[{}] writingSlow=[{}] timedOut=[{}] {}", threadId, readingSlow, writingSlow, timedOut, message);
+            logger.debug("[{}] Closing client connection [{}] due to inactivity", threadId, worker.getSocket().getRemoteSocketAddress());
 
             StringBuilder threadDump = new StringBuilder();
             for (Map.Entry<Thread, StackTraceElement[]> entry : Thread.getAllStackTraces().entrySet()) {
@@ -242,7 +254,7 @@ public void run() {
             }
           } catch (IOException e) {
             // Log but ignore
-            logger.debug("Unable to close connection to client. [{}]", e);
+            logger.debug(String.format("[%s] Unable to close connection to client. [{}]", threadId), e);
           }
         }
 
diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
index e5de47b..7043744 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -74,7 +74,7 @@ public HTTPWorker(Socket socket, HTTPServerConfiguration configuration, Instrume
     this.buffers = new HTTPBuffers(configuration);
     this.logger = configuration.getLoggerFactory().getLogger(HTTPWorker.class);
     this.state = State.Read;
-    logger.debug("Starting HTTP worker virtual thread");
+    logger.debug("[{}] Starting HTTP worker virtual thread", Thread.currentThread().threadId());
   }
 
   public Socket getSocket() {
@@ -91,7 +91,7 @@ public void run() {
       }
 
       while (true) {
-        logger.debug("Running HTTP worker and preparing to read preamble");
+        logger.trace("[{}] Running HTTP worker and preparing to read preamble", Thread.currentThread().threadId());
         var request = new HTTPRequest(configuration.getContextPath(), configuration.getMultipartBufferSize(),
             listener.getCertificate() != null ? "https" : "http", listener.getPort(), socket.getInetAddress().getHostAddress());
 
@@ -129,58 +129,61 @@ public void run() {
           keepAlive = false;
         }
 
-        logger.debug("Calling the handler.");
         var handler = configuration.getHandler();
         state = State.Process; // Transition to processing
+        logger.debug("[{}] Set state [{}], calling the handler", Thread.currentThread().threadId(), state);
         handler.handle(request, response);
         response.close();
-        logger.debug("Handler completed successfully.");
+        logger.trace("[{}] Handler completed successfully", Thread.currentThread().threadId());
 
         // Since the Handler can change the Keep-Alive state, we use the response here
         if (!keepAlive) {
-          logger.debug("Closing because no Keep-Alive.");
+          logger.debug("[{}] Close socket. No Keep-Alive", Thread.currentThread().threadId());
           close(Result.Success, response);
           break;
         }
 
         // Transition to Keep-Alive state and reset the SO timeout
-        logger.debug("Keeping things alive.");
         state = State.KeepAlive;
+        logger.debug("[{}] Set state [{}]", Thread.currentThread().threadId(), state);
         socket.setSoTimeout((int) configuration.getKeepAliveTimeoutDuration().toMillis());
 
         // Purge the extra bytes in case the handler didn't read everything
         int purged = httpInputStream.purge();
-        logger.trace("Purged [{}] bytes.", purged);
+        if (purged > 0) {
+          logger.debug("[{}] Purged [{}] bytes.", purged, Thread.currentThread().threadId());
+        }
+
       }
     } catch (SocketTimeoutException | ConnectionClosedException e) {
       // This might be a read timeout or a Keep-Alive timeout. The failure state is based on that flag
       close(keepAlive ? Result.Success : Result.Failure, response);
 
       if (keepAlive) {
-        logger.debug("Closing because the Keep-Alive expired.", e);
+        logger.debug(String.format("[%s] Closing socket. Keep-Alive expired.", Thread.currentThread().threadId()), e);
       } else {
-        logger.debug("Closing because socket timed out.");
+        logger.debug("[{}] Closing socket. Socket timed out.", Thread.currentThread().threadId());
       }
     } catch (ParseException pe) {
       if (instrumenter != null) {
         instrumenter.badRequest();
       }
 
-      logger.debug("Closing because of a bad request.");
+      logger.debug("[{}] Closing socket. Bad request. Reason [{}]", Thread.currentThread().threadId(), pe.getMessage());
       close(Result.Failure, response);
     } catch (SocketException e) {
       // This should only happen when the server is shutdown and this thread is waiting to read or write. In that case, this will throw a
       // SocketException and the thread will be interrupted. Since the server is being shutdown, we should let the client know.
       if (Thread.currentThread().isInterrupted()) {
-        logger.debug("Closing because server was shutdown.");
+        logger.debug("[{}] Closing socket. Server is shutting down.", Thread.currentThread().threadId());
         close(Result.Success, response);
       }
     } catch (IOException io) {
-      logger.debug("An IO exception was thrown during processing. These are pretty common.", io);
+      logger.debug(String.format("[%s] Closing socket. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId()), io);
       close(Result.Failure, response);
     } catch (Throwable t) {
       // Log the error and signal a failure
-      logger.error("HTTP worker threw an exception while processing a request.", t);
+      logger.error(String.format("[%s] Closing socket. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId()), t);
       close(Result.Failure, response);
     } finally {
       if (instrumenter != null) {
@@ -209,7 +212,7 @@ private void close(Result result, HTTPResponse response) {
 
       socket.close();
     } catch (IOException e) {
-      logger.debug("Could not close the connection because the socket threw an exception.", e);
+      logger.debug(String.format("[%s] Could not close the connection because the socket threw an exception.", Thread.currentThread().threadId()), e);
     }
   }
 
diff --git a/src/main/java/io/fusionauth/http/server/io/Throughput.java b/src/main/java/io/fusionauth/http/server/io/Throughput.java
index 3d4150f..87bbe4f 100644
--- a/src/main/java/io/fusionauth/http/server/io/Throughput.java
+++ b/src/main/java/io/fusionauth/http/server/io/Throughput.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2024-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -73,16 +73,19 @@ public synchronized long readThroughput(long now) {
       return Long.MAX_VALUE;
     }
 
+    // No bytes have been written
     if (numberOfBytesWritten == 0) {
       long millis = now - firstReadInstant;
+      // We are within the read delay and do not yet have enough data to make a meaningful calculation for actual throughput.
       if (millis < readThroughputDelay) {
         return Long.MAX_VALUE;
       }
 
-      double result = ((double) numberOfBytesRead / (double) millis) * 1_000;
-      return Math.round(result);
+      // Always zero
+      return numberOfBytesWritten;
     }
 
+    // The number of bytes read in seconds
     double result = ((double) numberOfBytesRead / (double) (lastReadInstant - firstReadInstant)) * 1_000;
     return Math.round(result);
   }
diff --git a/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java b/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java
index 282dc70..3326c06 100644
--- a/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java
+++ b/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java
@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Files;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -168,14 +169,14 @@ public Object[][] parts() {
         ------WebKitFormBoundaryTWfMVJErBoLURJIe--""";
     Object[][] invocations = new Object[body.length() - 1][];
     for (int i = 1; i < body.length(); i++) {
-      invocations[i - 1] = new Object[]{i, new byte[][]{body.substring(0, i).getBytes(), body.substring(i).getBytes()}};
+      invocations[i - 1] = new Object[]{i, new Parts(new byte[][]{body.substring(0, i).getBytes(), body.substring(i).getBytes()})};
     }
     return invocations;
   }
 
   @Test(dataProvider = "parts")
-  public void separateParts(@SuppressWarnings("unused") int index, byte[][] parts) throws IOException {
-    PartInputStream is = new PartInputStream(parts);
+  public void separateParts(@SuppressWarnings("unused") int index, Parts parts) throws IOException {
+    PartInputStream is = new PartInputStream(parts.parts);
     Map<String, List<String>> parameters = new HashMap<>();
     List<FileInfo> files = new LinkedList<>();
     MultipartStream stream = new MultipartStream(is, "----WebKitFormBoundaryTWfMVJErBoLURJIe".getBytes(), 1024);
@@ -225,4 +226,20 @@ public int read() {
       throw new UnsupportedOperationException();
     }
   }
+
+  public static class Parts {
+    public byte[][] parts;
+
+    public Parts(byte[][] parts) {
+      this.parts = parts;
+    }
+
+    public String toString() {
+      List<String> result = new ArrayList<>();
+      for (byte[] part : parts) {
+        result.add("" + part.length);
+      }
+      return "{" + String.join(",", result) + "}";
+    }
+  }
 }

From 55a2abab3bf10a0042a633705764ac16632b4fde Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Tue, 13 May 2025 17:23:00 -0600
Subject: [PATCH 02/15] working

---
 .../java/io/fusionauth/http/server/internal/HTTPWorker.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
index 7043744..812b798 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
@@ -131,7 +131,7 @@ public void run() {
 
         var handler = configuration.getHandler();
         state = State.Process; // Transition to processing
-        logger.debug("[{}] Set state [{}], calling the handler", Thread.currentThread().threadId(), state);
+        logger.debug("[{}] Set state [{}]. Call the request handler.", Thread.currentThread().threadId(), state);
         handler.handle(request, response);
         response.close();
         logger.trace("[{}] Handler completed successfully", Thread.currentThread().threadId());

From 5a2c0594f93de5f819b2088d294022a80ebb7d5f Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Tue, 13 May 2025 17:32:06 -0600
Subject: [PATCH 03/15] working

---
 .../io/fusionauth/http/server/internal/HTTPServerThread.java    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
index df8d4b4..4e258c3 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
@@ -184,7 +184,7 @@ public void run() {
           long readThroughput = -1;
           long writeThroughput = -1;
 
-          String badClientReason = "[" + threadId + "] Checking worker in state [" + state + "]";
+          String badClientReason = "[" + threadId + "] Check worker in state [" + state + "]";
           if (state == HTTPWorker.State.Read) {
             // Here the SO_TIMEOUT set above or the Keep-Alive timeout in HTTPWorker will dictate if the socket has timed out. This prevents slow readers
             // or network issues where the client reads 1 byte per timeout value (i.e. 1 byte per 2 seconds or something like that)

From 981f84ed41ede33ea3718b343ce6010f0e1cd1a3 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Tue, 13 May 2025 17:32:35 -0600
Subject: [PATCH 04/15] working

---
 build.savant | 2 +-
 pom.xml      | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/build.savant b/build.savant
index da1bc59..8350d52 100644
--- a/build.savant
+++ b/build.savant
@@ -17,7 +17,7 @@ jackson5Version = "3.0.1"
 restifyVersion = "4.2.1"
 testngVersion = "7.10.2"
 
-project(group: "io.fusionauth", name: "java-http", version: "0.4.0-RC.5", licenses: ["ApacheV2_0"]) {
+project(group: "io.fusionauth", name: "java-http", version: "0.4.0-RC.6", licenses: ["ApacheV2_0"]) {
   workflow {
     fetch {
       // Dependency resolution order:
diff --git a/pom.xml b/pom.xml
index 17fcbb4..00bb874 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,7 +2,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>io.fusionauth</groupId>
   <artifactId>java-http</artifactId>
-  <version>0.4.0-RC.4</version>
+  <version>0.4.0-RC.6</version>
   <packaging>jar</packaging>
 
   <name>Java HTTP library (client and server)</name>

From 22dc35cfb1649ce2c1e16132d556a9a058ea8a47 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Tue, 13 May 2025 18:13:11 -0600
Subject: [PATCH 05/15] working

---
 .../fusionauth/http/server/HTTPResponse.java  | 42 ++++++++++---------
 1 file changed, 22 insertions(+), 20 deletions(-)

diff --git a/src/main/java/io/fusionauth/http/server/HTTPResponse.java b/src/main/java/io/fusionauth/http/server/HTTPResponse.java
index 63c9672..14ec5fe 100644
--- a/src/main/java/io/fusionauth/http/server/HTTPResponse.java
+++ b/src/main/java/io/fusionauth/http/server/HTTPResponse.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021-2022, FusionAuth, All Rights Reserved
+ * Copyright (c) 2021-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -147,23 +147,6 @@ public Long getContentLength() {
     return null;
   }
 
-  /**
-   * Hard resets this response if it hasn't been committed yet. If the response has been committed back to the client, this throws up.
-   */
-  public void reset() {
-    if (outputStream.isCommitted()) {
-      throw new IllegalStateException("The HTTPResponse can't be reset after it has been committed, meaning at least one byte was written back to the client.");
-    }
-
-    cookies.clear();
-    headers.clear();
-    exception = null;
-    outputStream.reset();
-    status = 200;
-    statusMessage = null;
-    writer = null;
-  }
-
   public void setContentLength(long length) {
     setHeader(Headers.ContentLength, Long.toString(length));
   }
@@ -272,8 +255,10 @@ public void setCompress(boolean compress) {
    *     this method does.
    */
   public boolean isKeepAlive() {
-    String value = getHeader(Headers.Connection);
-    return value == null || value.equalsIgnoreCase(Connections.KeepAlive);
+    // TODO : Daniel Review : Should this have the same logic as HTTPRequest.isKeepAlive?
+    //        - The request says if null or != "close" then it is keep alive.
+    String connection = getHeader(Headers.Connection);
+    return connection == null || connection.equalsIgnoreCase(Connections.KeepAlive);
   }
 
   public void removeCookie(String name) {
@@ -291,6 +276,23 @@ public void removeHeader(String name) {
     }
   }
 
+  /**
+   * Hard resets this response if it hasn't been committed yet. If the response has been committed back to the client, this throws up.
+   */
+  public void reset() {
+    if (outputStream.isCommitted()) {
+      throw new IllegalStateException("The HTTPResponse can't be reset after it has been committed, meaning at least one byte was written back to the client.");
+    }
+
+    cookies.clear();
+    headers.clear();
+    exception = null;
+    outputStream.reset();
+    status = 200;
+    statusMessage = null;
+    writer = null;
+  }
+
   public void sendRedirect(String uri) {
     setHeader(Headers.Location, uri);
     status = Status.MovedTemporarily;

From f128fbccac9cbf996f73f9b893cd3105dfa3ebd3 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Wed, 14 May 2025 21:40:35 -0600
Subject: [PATCH 06/15] Add logging, and tests

---
 .../io/fusionauth/http/ParseException.java    |  15 +-
 .../http/server/internal/HTTPWorker.java      | 141 +++++++---
 .../http/util/RequestPreambleState.java       |   8 +-
 .../java/io/fusionauth/http/BaseTest.java     |  19 +-
 .../java/io/fusionauth/http/CoreTest.java     |   5 +-
 .../io/fusionauth/http/HTTP11SpecTest.java    | 262 ++++++++++++++++++
 .../io/fusionauth/http/ServerRunTest.java     |  41 +++
 7 files changed, 445 insertions(+), 46 deletions(-)
 create mode 100644 src/test/java/io/fusionauth/http/HTTP11SpecTest.java
 create mode 100644 src/test/java/io/fusionauth/http/ServerRunTest.java

diff --git a/src/main/java/io/fusionauth/http/ParseException.java b/src/main/java/io/fusionauth/http/ParseException.java
index 2f5d5c8..f42e61e 100644
--- a/src/main/java/io/fusionauth/http/ParseException.java
+++ b/src/main/java/io/fusionauth/http/ParseException.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,18 +21,23 @@
  * @author Brian Pontarelli
  */
 public class ParseException extends RuntimeException {
+  private final String state;
+
   public ParseException() {
+    this.state = null;
   }
 
   public ParseException(String message) {
     super(message);
+    this.state = null;
   }
 
-  public ParseException(String message, Throwable cause) {
-    super(message, cause);
+  public ParseException(String message, String state) {
+    super(message);
+    this.state = state;
   }
 
-  public ParseException(Throwable cause) {
-    super(cause);
+  public String getState() {
+    return state;
   }
 }
diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
index 812b798..ffe4072 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
@@ -95,16 +95,33 @@ public void run() {
         var request = new HTTPRequest(configuration.getContextPath(), configuration.getMultipartBufferSize(),
             listener.getCertificate() != null ? "https" : "http", listener.getPort(), socket.getInetAddress().getHostAddress());
 
+        // Set up the output stream so that if we fail we have the opportunity to write a response that contains a status code.
+        var throughputOutputStream = new ThroughputOutputStream(socket.getOutputStream(), throughput);
+        response = new HTTPResponse();
+
+        HTTPOutputStream outputStream = new HTTPOutputStream(configuration, request.getAcceptEncodings(), response, throughputOutputStream, buffers, () -> state = State.Write);
+        response.setOutputStream(outputStream);
+
         var inputStream = new ThroughputInputStream(socket.getInputStream(), throughput);
         var bodyBytes = HTTPTools.parseRequestPreamble(inputStream, request, buffers.requestBuffer(), instrumenter, () -> state = State.Read);
         var httpInputStream = new HTTPInputStream(configuration, request, inputStream, bodyBytes);
         request.setInputStream(httpInputStream);
 
-        var throughputOutputStream = new ThroughputOutputStream(socket.getOutputStream(), throughput);
-        response = new HTTPResponse();
+        // Set the Connection response header as soon as possible
+        if (request.isKeepAlive()) {
+          response.setHeader(Headers.Connection, Connections.KeepAlive);
+          keepAlive = true;
+        } else {
+          response.setHeader(Headers.Connection, Connections.Close);
+          keepAlive = false;
+        }
 
-        HTTPOutputStream outputStream = new HTTPOutputStream(configuration, request.getAcceptEncodings(), response, throughputOutputStream, buffers, () -> state = State.Write);
-        response.setOutputStream(outputStream);
+        // Ensure the preamble is valid
+        Integer status = validatePreamble(request);
+        if (status != null) {
+          close(status, response);
+          return;
+        }
 
         // Handle the "expect" response
         String expect = request.getHeader(Headers.Expect);
@@ -113,7 +130,7 @@ public void run() {
 
           // If the "expect" wasn't accepted, close the socket and exit
           if (!expectContinue(request)) {
-            close(Result.Success, response);
+            close(Status.OK, response);
             return;
           }
 
@@ -121,14 +138,6 @@ public void run() {
           state = State.Read;
         }
 
-        if (request.isKeepAlive()) {
-          response.setHeader(Headers.Connection, Connections.KeepAlive);
-          keepAlive = true;
-        } else {
-          response.setHeader(Headers.Connection, Connections.Close);
-          keepAlive = false;
-        }
-
         var handler = configuration.getHandler();
         state = State.Process; // Transition to processing
         logger.debug("[{}] Set state [{}]. Call the request handler.", Thread.currentThread().threadId(), state);
@@ -139,7 +148,7 @@ public void run() {
         // Since the Handler can change the Keep-Alive state, we use the response here
         if (!keepAlive) {
           logger.debug("[{}] Close socket. No Keep-Alive", Thread.currentThread().threadId());
-          close(Result.Success, response);
+          close(Status.OK, response);
           break;
         }
 
@@ -157,34 +166,39 @@ public void run() {
       }
     } catch (SocketTimeoutException | ConnectionClosedException e) {
       // This might be a read timeout or a Keep-Alive timeout. The failure state is based on that flag
-      close(keepAlive ? Result.Success : Result.Failure, response);
+      var status = keepAlive ? Status.OK : Status.InternalServerError; // 200 or 500
+      close(status, response);
 
       if (keepAlive) {
-        logger.debug(String.format("[%s] Closing socket. Keep-Alive expired.", Thread.currentThread().threadId()), e);
+        logger.debug(String.format("[%s] Closing socket with status [%d]. Keep-Alive expired.", Thread.currentThread().threadId(), status), e);
       } else {
-        logger.debug("[{}] Closing socket. Socket timed out.", Thread.currentThread().threadId());
+        logger.debug("[{}] Closing socket with status [{}]. Socket timed out.", Thread.currentThread().threadId(), status);
       }
     } catch (ParseException pe) {
       if (instrumenter != null) {
         instrumenter.badRequest();
       }
 
-      logger.debug("[{}] Closing socket. Bad request. Reason [{}]", Thread.currentThread().threadId(), pe.getMessage());
-      close(Result.Failure, response);
+      int status = 400;
+      logger.debug("[{}] Closing socket with status [{}]. Bad request, failed to parse request. Reason [{}] Parser state [{}]", Thread.currentThread().threadId(), status, pe.getMessage(), pe.getState());
+      close(status, response);
     } catch (SocketException e) {
       // This should only happen when the server is shutdown and this thread is waiting to read or write. In that case, this will throw a
       // SocketException and the thread will be interrupted. Since the server is being shutdown, we should let the client know.
       if (Thread.currentThread().isInterrupted()) {
-        logger.debug("[{}] Closing socket. Server is shutting down.", Thread.currentThread().threadId());
-        close(Result.Success, response);
+        var status = Status.OK;
+        logger.debug("[{}] Closing socket with status [{}]. Server is shutting down.", Thread.currentThread().threadId(), status);
+        close(status, response);
       }
     } catch (IOException io) {
-      logger.debug(String.format("[%s] Closing socket. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId()), io);
-      close(Result.Failure, response);
+      var status = Status.InternalServerError;
+      logger.debug(String.format("[%s] Closing socket with status [%d]. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId(), status), io);
+      close(status, response);
     } catch (Throwable t) {
       // Log the error and signal a failure
-      logger.error(String.format("[%s] Closing socket. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId()), t);
-      close(Result.Failure, response);
+      var status = Status.InternalServerError;
+      logger.error(String.format("[%s] Closing socket with status [%d]. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId(), status), t);
+      close(status, response);
     } finally {
       if (instrumenter != null) {
         instrumenter.threadExited();
@@ -196,16 +210,21 @@ public State state() {
     return state;
   }
 
-  private void close(Result result, HTTPResponse response) {
-    if (result == Result.Failure && instrumenter != null) {
+  private void close(int status, HTTPResponse response) {
+    boolean error = status != 200;
+    if (error && instrumenter != null) {
       instrumenter.connectionClosed();
     }
 
     try {
       // If the conditions are perfect, we can still write back a 500
-      if (result == Result.Failure && response != null && !response.isCommitted()) {
-        response.reset();
-        response.setStatus(500);
+      if (error && response != null && !response.isCommitted()) {
+        // Note that reset() clears the Connection response header.
+//        response.reset();
+        response.setStatus(status);
+        // TODO: Daniel : Should we always be sending back a Connection: close on an error such as this?
+        //      - I think this is correct since we are calling socket.close() below?
+        response.setHeader(Headers.Connection, Connections.Close);
         response.setContentLength(0L);
         response.close();
       }
@@ -231,6 +250,57 @@ private boolean expectContinue(HTTPRequest request) throws IOException {
     return expectResponse.getStatus() == 100;
   }
 
+  private Integer validatePreamble(HTTPRequest request) {
+    // Validate protocol. Only HTTP/1.1 is supported
+    String protocol = request.getProtocol();
+    if (protocol == null) {
+      if (logger.isTraceEnabled()) {
+        logger.trace("Invalid request. Missing HTTP Protocol");
+      }
+      return Status.BadRequest;
+    }
+
+    if (!protocol.startsWith("HTTP/")) {
+      if (logger.isTraceEnabled()) {
+        logger.trace("Invalid request. Invalid protocol [{}]. Supported versions [HTTP/1.1].", protocol);
+      }
+      return Status.BadRequest;
+    }
+
+    if (!"HTTP/1.1".equals(protocol)) {
+      if (logger.isTraceEnabled()) {
+        logger.trace("Invalid request. Unsupported HTTP version [{}]. Supported versions [HTTP/1.1].", protocol);
+      }
+      return Status.HTTPVersionNotSupported;
+    }
+
+    // Host header is required
+    var host = request.getRawHost();
+    if (host == null) {
+      logger.trace("Invalid request. Missing Host header.");
+      return Status.BadRequest;
+    }
+
+    var hostHeaders = request.getHeaders(Headers.Host);
+    if (hostHeaders.size() != 1) {
+      if (logger.isTraceEnabled()) {
+        logger.trace("Invalid request. Duplicate Host headers. [{}]", String.join(", ", hostHeaders));
+      }
+      return Status.BadRequest;
+    }
+
+    // Content-Length cannot be negative
+    var contentLength = request.getContentLength();
+    if (contentLength != null && contentLength < 0) {
+      if (logger.isTraceEnabled()) {
+        logger.trace("Invalid request. The Content-Length cannot be negative. [{}]", contentLength);
+      }
+      return Status.BadRequest;
+    }
+
+    return null;
+  }
+
   public enum State {
     Read,
     Process,
@@ -238,8 +308,13 @@ public enum State {
     KeepAlive
   }
 
-  private enum Result {
-    Failure,
-    Success
+  private static class Status {
+    public static final int BadRequest = 400;
+
+    public static final int HTTPVersionNotSupported = 505;
+
+    public static final int InternalServerError = 500;
+
+    public static final int OK = 200;
   }
 }
diff --git a/src/main/java/io/fusionauth/http/util/RequestPreambleState.java b/src/main/java/io/fusionauth/http/util/RequestPreambleState.java
index 532a659..0807778 100644
--- a/src/main/java/io/fusionauth/http/util/RequestPreambleState.java
+++ b/src/main/java/io/fusionauth/http/util/RequestPreambleState.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -98,7 +98,9 @@ public boolean store() {
   RequestProtocol {
     @Override
     public RequestPreambleState next(byte ch) {
-      if (ch == 'H' || ch == 'T' || ch == 'P' || ch == '/' || ch == '1' || ch == '.') {
+      // While this server only supports HTTP/1.1, allow the request protocol to be parsed for any valid version.
+      // - The supported version will be validated elsewhere.
+      if (ch == 'H' || ch == 'T' || ch == 'P' || ch == '/' || ch == '.' || (ch >= '0' && ch <= '9')) {
         return RequestProtocol;
       } else if (ch == '\r') {
         return RequestCR;
@@ -268,6 +270,6 @@ public boolean store() {
   public abstract boolean store();
 
   ParseException makeParseException(char ch) {
-    return new ParseException("Invalid character [" + ch + "] in state [" + this + "]");
+    return new ParseException("Invalid character [" + ch + "] in state [" + this + "]", this.name());
   }
 }
diff --git a/src/test/java/io/fusionauth/http/BaseTest.java b/src/test/java/io/fusionauth/http/BaseTest.java
index d31914e..db8922f 100644
--- a/src/test/java/io/fusionauth/http/BaseTest.java
+++ b/src/test/java/io/fusionauth/http/BaseTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -325,8 +325,19 @@ public void sendBadRequest(String message) {
 
       // Sockets are pretty resilient, so this will be closed by the server, but we'll just see that close are zero bytes read. If we were
       // to continue writing above, then that likely would throw an exception because the pipe would be broken
+
+      // TODO : Daniel :
+      //        In many cases we still return a 400 to the client when a request is invalid. So there will be bytes to read.
+      //        We could require the caller to provide and expect status code, and or response body.
+      //        Currently, this is just used by one test, so I have hard coded the expected response.
+
       byte[] buffer = is.readAllBytes();
-      assertEquals(buffer.length, 0);
+      assertEquals(new String(buffer), """
+          HTTP/1.1 400 \r
+          connection: close\r
+          content-length: 0\r
+          \r
+          """);
     } catch (Exception e) {
       fail(e.getMessage());
     }
@@ -392,12 +403,12 @@ public void onTestFailure(ITestResult result) {
       }
 
       System.out.println("""
-
+          
           Test failure
           -----------------
           Exception: {{exception}}
           Message: {{message}}
-
+          
           Stack traces (client side):
           {{threadDump}}
           -----------------
diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index f0ccea6..6ec3cf3 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -99,11 +99,14 @@ public void badPreambleButReset() throws Exception {
 
     var instrumenter = new CountingInstrumenter();
     try (var client = HttpClient.newHttpClient(); var ignore = makeServer("http", handler, instrumenter).start()) {
+      // Invalid request, missing Host header
+      // - This should cause the socket to be reset
       sendBadRequest("""
           GET / HTTP/1.1\r
           X-Bad-Header: Bad-Header\r\r
           """);
 
+      // TODO : Daniel : Review : What is this test doing? If I comment out reset() in HTTPWorker.close() the test still passes.
       URI uri = makeURI("http", "");
       HttpRequest request = HttpRequest.newBuilder()
                                        .uri(uri)
diff --git a/src/test/java/io/fusionauth/http/HTTP11SpecTest.java b/src/test/java/io/fusionauth/http/HTTP11SpecTest.java
new file mode 100644
index 0000000..22fbae1
--- /dev/null
+++ b/src/test/java/io/fusionauth/http/HTTP11SpecTest.java
@@ -0,0 +1,262 @@
+/*
+ * Copyright (c) 2025, FusionAuth, All Rights Reserved
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific
+ * language governing permissions and limitations under the License.
+ */
+package io.fusionauth.http;
+
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+
+import io.fusionauth.http.server.HTTPServer;
+import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Tests the HTTP server for HTTP 1.1. spec compliance.
+ *
+ * @author Daniel DeGroff
+ */
+public class HTTP11SpecTest extends BaseTest {
+  @Test
+  public void duplicate_host_header() throws Exception {
+    withRequest("""
+        GET / HTTP/1.1\r
+        Host: cyberdyne-systems.com\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+  }
+
+  @Test
+  public void host_required() throws Exception {
+    //Host header is required, return 400 if not provided
+    withRequest("""
+        GET / HTTP/1.1\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Ensure that X-Forwarded-Host doesn't count for the Host header
+    withRequest("""
+        GET / HTTP/1.1\r
+        X-Forwarded-Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+  }
+
+  @Test
+  public void invalid_content_length() throws Exception {
+    // Too large, we will take a NumberFormatException and set this value to null in the request.
+    // - So as it is written, we won't return the user an error, but we will assume no body is present.
+    withRequest("""
+        GET / HTTP/1.1\r
+        Host: cyberdyne-systems.com\r
+        Content-Length: 9223372036854775808\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 200 \r
+        connection: keep-alive\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Negative
+    withRequest("""
+        GET / HTTP/1.1\r
+        Host: cyberdyne-systems.com\r
+        Content-Length: -1\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Max Long - but negative
+    withRequest("""
+        GET / HTTP/1.1\r
+        Host: cyberdyne-systems.com\r
+        Content-Length: -9223372036854775807\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+  }
+
+  @Test
+  public void invalid_version() throws Exception {
+    // We should be returning 505 unless we see HTTP/1.1
+
+    // Invalid: HTTP/1
+    withRequest("""
+        GET / HTTP/1\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse(
+        """
+            HTTP/1.1 505 \r
+            connection: close\r
+            content-length: 0\r
+            \r
+            """);
+
+    // Invalid: HTTP/1.
+    withRequest("""
+        GET / HTTP/1.\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 505 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Invalid: HTTP/1.0
+    withRequest("""
+        GET / HTTP/1.0\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 505 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Invalid: HTTP/1.2
+    withRequest("""
+        GET / HTTP/1.2\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 505 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Invalid: HTTP/9.9
+    withRequest("""
+        GET / HTTP/9.9\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 505 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+  }
+
+  @Test
+  public void mangled_version() throws Exception {
+    // HTTP reversed, does not begin with HTTP/
+    withRequest("""
+        GET / PTTH/1.1\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+
+    // Repeated allowed characters, does not begin with HTTP/
+    withRequest("""
+        GET / HHHH/1.1\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+  }
+
+  @Test
+  public void missing_protocol() throws Exception {
+    withRequest("""
+        GET /\r
+        Host: cyberdyne-systems.com\r
+        \r
+        """
+    ).expectResponse("""
+        HTTP/1.1 400 \r
+        connection: close\r
+        content-length: 0\r
+        \r
+        """);
+  }
+
+  private void assertResponse(String request, String response) throws Exception {
+    try (HTTPServer ignore = makeServer("http", (req, res) -> res.setStatus(200)).start();
+         Socket sock = makeClientSocket("http")) {
+      var os = sock.getOutputStream();
+      os.write(request.getBytes(StandardCharsets.UTF_8));
+      os.flush();
+      var resp = new String(sock.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
+      assertEquals(resp, response, "HTTP request:\n" + request + "\n");
+    }
+  }
+
+  private Builder withRequest(String request) {
+    return new Builder(request);
+  }
+
+  private class Builder {
+    public String request;
+
+    public Builder(String request) {
+      this.request = request;
+    }
+
+    public void expectResponse(String response) throws Exception {
+      assertResponse(request, response);
+    }
+  }
+}
diff --git a/src/test/java/io/fusionauth/http/ServerRunTest.java b/src/test/java/io/fusionauth/http/ServerRunTest.java
new file mode 100644
index 0000000..3d6e790
--- /dev/null
+++ b/src/test/java/io/fusionauth/http/ServerRunTest.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2025, FusionAuth, All Rights Reserved
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific
+ * language governing permissions and limitations under the License.
+ */
+package io.fusionauth.http;
+
+import java.time.Duration;
+
+import io.fusionauth.http.server.HTTPServer;
+import org.testng.annotations.Test;
+
+/**
+ * Run a test so we can run things against it externally. Should probably move this to a test target instead?
+ *
+ * @author Daniel DeGroff
+ */
+public class ServerRunTest extends BaseTest {
+  // Enable to just run a server to test against externally. Could move this to a build target?
+  @Test(enabled = false)
+  public void server_run() throws Exception {
+    Duration runtime = Duration.ofHours(1);
+    try (HTTPServer server = makeServer("http", (req, res) -> res.setStatus(200), null).start()) {
+      System.out.println("Server started on port: " +
+                         server.configuration().getListeners().getFirst().getPort() +
+                         "\n - Run for [" + runtime.toMinutes() + "] minutes");
+      Thread.sleep(runtime.toMillis());
+      System.out.println(" - Exit");
+    }
+  }
+}

From dfdc104e7b12cc27d9fd455fd795e96aeb9522a5 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Wed, 14 May 2025 21:57:25 -0600
Subject: [PATCH 07/15] Tests

---
 .../http/server/internal/HTTPWorker.java         |  2 +-
 src/test/java/io/fusionauth/http/CoreTest.java   | 16 +++++++++++++---
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
index ffe4072..95007ac 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
@@ -220,7 +220,7 @@ private void close(int status, HTTPResponse response) {
       // If the conditions are perfect, we can still write back a 500
       if (error && response != null && !response.isCommitted()) {
         // Note that reset() clears the Connection response header.
-//        response.reset();
+        response.reset();
         response.setStatus(status);
         // TODO: Daniel : Should we always be sending back a Connection: close on an error such as this?
         //      - I think this is correct since we are calling socket.close() below?
diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index 6ec3cf3..823b67c 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -62,11 +62,21 @@
  */
 @SuppressWarnings({"UastIncorrectHttpHeaderInspection", "OptionalGetWithoutIsPresent", "ThrowablePrintedToSystemOut"})
 public class CoreTest extends BaseTest {
-  public static final String ExpectedResponse = "{\"version\":\"42\"}";
+  public static final String ExpectedResponse = """
+      {
+        "version": "42"
+      }
+      """
+      .replaceAll("\\s", "");
 
-  public static final String LongString = "12345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890".repeat(64);
+  // This string is 16,640 characters long
+  public static final String LongString = "1234567890".repeat(1_664);
 
-  public static final String RequestBody = "{\"message\":\"Hello World\"";
+  public static final String RequestBody = """
+      {
+        "message": "Hello World"
+      }
+      """;
 
   @Test(dataProvider = "schemes")
   public void badLanguage(String scheme) throws Exception {

From 8c8c3ec4a6627e917c331b0509f4536548d725ad Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Wed, 14 May 2025 22:22:59 -0600
Subject: [PATCH 08/15] Tests

---
 .../java/io/fusionauth/http/CoreTest.java     | 24 ++++++++++++++-----
 1 file changed, 18 insertions(+), 6 deletions(-)

diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index 823b67c..c4e123b 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -589,19 +589,28 @@ public void serverClosesSockets(String scheme) {
   public void serverTimeout() throws Exception {
     // This test simulates if the server has a long-running thread that doesn't write fast enough
     HTTPHandler handler = (req, res) -> {
-      System.out.println("Handling");
+      System.out.println("Handling ... (slowly)");
+      // Note that if you comment out the sleep, the test should fail.
       sleep(4_000L);
       res.setStatus(200);
-      res.setContentLength(0L);
+      var body = "I'm slow but I'm good.".getBytes(StandardCharsets.UTF_8);
+      res.setContentLength(body.length);
+      res.getOutputStream().write(body);
       res.getOutputStream().close();
       System.out.println("Closed");
     };
 
+    // TODO : Daniel Review : what are we trying to test here, this test isn't doing what we think.
+    //        It seems we are writing to the server as a client, and the server is slow to write back.
+    //        But this would be a client timeout and not a server timeout?
+    //       Seems like we should use withProcessingTimeoutDuration instead of withInitialReadTimeout?
     var instrumenter = new CountingInstrumenter();
-    try (var ignore = makeServer("http", handler, instrumenter).withInitialReadTimeout(Duration.ofSeconds(1)).start(); Socket socket = new Socket("127.0.0.1", 4242)) {
+    try (var ignore = makeServer("http", handler, instrumenter).withProcessingTimeoutDuration(Duration.ofSeconds(1)).start(); Socket socket = new Socket("127.0.0.1", 4242)) {
       var out = socket.getOutputStream();
+      // 1. Write a body to the server
       out.write("""
           GET / HTTP/1.1\r
+          Host: localhost:42\r
           Connection: close\r
           Content-Length: 4\r
           \r
@@ -609,9 +618,12 @@ public void serverTimeout() throws Exception {
           """.getBytes());
       out.flush();
 
-      var in = socket.getInputStream();
-      var body = in.readAllBytes();
-      assertEquals(body.length, 0, new String(body));
+      // 2. Read from the server, assuming you will receive a response.
+      //   However, the server is very slow, it is going to wait 4s before it writes the response.
+      //    - We have configured the server with a 1s read timeout.
+      //    - Expect that we will have received a SocketTimeoutException and as such the response will be empty.
+      var response = socket.getInputStream().readAllBytes();
+      assertEquals(response.length, 0, new String(response));
       assertEquals(instrumenter.getClosedConnections(), 1);
     }
   }

From 68366047f01cd9bced5f4dd4018374c92de4bb00 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Wed, 14 May 2025 22:38:18 -0600
Subject: [PATCH 09/15] Tests

---
 src/test/java/io/fusionauth/http/CoreTest.java | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index c4e123b..fee5b15 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -784,6 +784,7 @@ public void slowClient(String scheme) throws Exception {
     byte[] bytes = new byte[1024 * 1024 * 8];
     new SecureRandom().nextBytes(bytes);
 
+    // The server will write this back to the client
     HTTPHandler handler = (req, res) -> {
       res.setContentType("application/octet-stream");
       res.setContentLength(bytes.length);
@@ -798,10 +799,17 @@ public void slowClient(String scheme) throws Exception {
       out.close();
     };
 
+    // TODO : Daniel Review : What are we testing here? Is this testing the "reader too slow" which is the read throughput calculation?
+    //        Or the initial read timeout?
+    //        Neither were being changed from the defaults. Default bytes / second on the read is 16k.
+    //        The test comments below mention waiting for 2 seconds, so assuming that is referring to the initialReadTimeoutDuration?
+    //        This test is flaky, adding Connection: close on the client doesn't seem to help either.
+    //        It does seem to fail on http, but pass on https for some reason. Needs investigation.
     AtomicBoolean slept = new AtomicBoolean(false);
     try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler).start()) {
       URI uri = makeURI(scheme, "");
       client.send(
+          // Don't keep this connection open because the timeouts aren't the same on a keep alive.
           HttpRequest.newBuilder().uri(uri).GET().build(),
           r -> BodySubscribers.ofByteArrayConsumer(optional -> {
             byte[] actual = optional.orElse(null);

From c709bbb4071758e9f40c74248c6ba598444cb0a6 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Wed, 14 May 2025 22:45:13 -0600
Subject: [PATCH 10/15] Tests

---
 src/test/java/io/fusionauth/http/CoreTest.java | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index fee5b15..caa8b6f 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -272,12 +272,16 @@ public void initialReadTimeout() {
     // This test simulates if the client doesn't send bytes for the initial timeout
     HTTPHandler handler = (req, res) -> fail("Should not be called");
 
+    // TODO : Daniel : Flaky test, need to review timeouts?, try invoking 5 or more times.
+    //                 Assuming that the socket gets re-used and that may be the difference in testing.
+    //        Maybe in the "timeouts" group we need to be sure to tear down the socket all the way in between tests?
     var instrumenter = new CountingInstrumenter();
     try (var ignore = makeServer("http", handler, instrumenter).withInitialReadTimeout(Duration.ofMillis(250)).start(); var socket = new Socket("127.0.0.1", 4242)) {
       var out = socket.getOutputStream();
       sleep(2_000);
       out.write("""
           GET / HTTP/1.1\r
+          Host: localhost:42\r
           Connection: close\r
           Content-Length: 4\r
           \r

From 309e8c3def65f5b7a5a8f424702d6da5415ddfdc Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Thu, 15 May 2025 15:41:51 -0600
Subject: [PATCH 11/15] Working

---
 .../http/server/internal/HTTPWorker.java      | 31 +++++++++++++++----
 .../java/io/fusionauth/http/ExpectTest.java   | 26 ++++++++++------
 2 files changed, 41 insertions(+), 16 deletions(-)

diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
index 95007ac..3a447f9 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java
@@ -25,6 +25,7 @@
 import io.fusionauth.http.HTTPValues;
 import io.fusionauth.http.HTTPValues.Connections;
 import io.fusionauth.http.HTTPValues.Headers;
+import io.fusionauth.http.HTTPValues.Protocols;
 import io.fusionauth.http.ParseException;
 import io.fusionauth.http.log.Logger;
 import io.fusionauth.http.server.HTTPHandler;
@@ -130,7 +131,7 @@ public void run() {
 
           // If the "expect" wasn't accepted, close the socket and exit
           if (!expectContinue(request)) {
-            close(Status.OK, response);
+            closeSocketOnly();
             return;
           }
 
@@ -148,7 +149,7 @@ public void run() {
         // Since the Handler can change the Keep-Alive state, we use the response here
         if (!keepAlive) {
           logger.debug("[{}] Close socket. No Keep-Alive", Thread.currentThread().threadId());
-          close(Status.OK, response);
+          closeSocketOnly();
           break;
         }
 
@@ -211,7 +212,7 @@ public State state() {
   }
 
   private void close(int status, HTTPResponse response) {
-    boolean error = status != 200;
+    boolean error = status > 299;
     if (error && instrumenter != null) {
       instrumenter.connectionClosed();
     }
@@ -235,18 +236,36 @@ private void close(int status, HTTPResponse response) {
     }
   }
 
+  private void closeSocketOnly() {
+    try {
+      socket.close();
+    } catch (IOException e) {
+      logger.debug(String.format("[%s] Could not close the connection because the socket threw an exception.", Thread.currentThread().threadId()), e);
+    }
+  }
+
   private boolean expectContinue(HTTPRequest request) throws IOException {
     var expectResponse = new HTTPResponse();
+    // Default to 100, the validator is optional. The HTTP response will default to 200.
+    expectResponse.setStatus(100);
+
+    // Invoke the optional validator
     var validator = configuration.getExpectValidator();
     if (validator != null) {
       validator.validate(request, expectResponse);
     }
 
+    // Ignore the status message if set by the validator. A status code of 100 should always have a message of Continue.
+    if (expectResponse.getStatus() == 100) {
+      expectResponse.setStatusMessage("Continue");
+    }
+
     // Write directly to the socket because the HTTPOutputStream does a lot of extra work that we don't want
     OutputStream out = socket.getOutputStream();
     HTTPTools.writeResponsePreamble(expectResponse, out);
     out.flush();
 
+    // A status code of 100 means go for it.
     return expectResponse.getStatus() == 100;
   }
 
@@ -262,14 +281,14 @@ private Integer validatePreamble(HTTPRequest request) {
 
     if (!protocol.startsWith("HTTP/")) {
       if (logger.isTraceEnabled()) {
-        logger.trace("Invalid request. Invalid protocol [{}]. Supported versions [HTTP/1.1].", protocol);
+        logger.trace("Invalid request. Invalid protocol [{}]. Supported versions [{}].", protocol, Protocols.HTTTP1_1);
       }
       return Status.BadRequest;
     }
 
-    if (!"HTTP/1.1".equals(protocol)) {
+    if (!Protocols.HTTTP1_1.equals(protocol)) {
       if (logger.isTraceEnabled()) {
-        logger.trace("Invalid request. Unsupported HTTP version [{}]. Supported versions [HTTP/1.1].", protocol);
+        logger.trace("Invalid request. Unsupported HTTP version [{}]. Supported versions [{}].", protocol, Protocols.HTTTP1_1);
       }
       return Status.HTTPVersionNotSupported;
     }
diff --git a/src/test/java/io/fusionauth/http/ExpectTest.java b/src/test/java/io/fusionauth/http/ExpectTest.java
index c28d289..c14c133 100644
--- a/src/test/java/io/fusionauth/http/ExpectTest.java
+++ b/src/test/java/io/fusionauth/http/ExpectTest.java
@@ -83,17 +83,23 @@ public void expect(String scheme) throws Exception {
       res.setStatusMessage("Continue");
     };
 
+    // Test w/ and w/out a custom expect validator. The default behavior should be to approve the payload.
     CountingInstrumenter instrumenter = new CountingInstrumenter();
-    try (HTTPServer ignore = makeServer(scheme, handler, instrumenter, validator).start(); var client = makeClient(scheme, null)) {
-      URI uri = makeURI(scheme, "");
-      var response = client.send(
-          HttpRequest.newBuilder().uri(uri).header(Headers.ContentType, "application/json").expectContinue(true).POST(BodyPublishers.ofString(RequestBody)).build(),
-          r -> BodySubscribers.ofString(StandardCharsets.UTF_8)
-      );
+    boolean[] validationOptions = {true, false};
+    for (boolean validation : validationOptions) {
+      ExpectValidator expectValidator = validation ? validator : null;
 
-      assertEquals(response.statusCode(), 200);
-      assertEquals(response.body(), ExpectedResponse);
-      assertTrue(validated.get());
+      try (HTTPServer ignore = makeServer(scheme, handler, instrumenter, expectValidator).start(); var client = makeClient(scheme, null)) {
+        URI uri = makeURI(scheme, "");
+        var response = client.send(
+            HttpRequest.newBuilder().uri(uri).header(Headers.ContentType, "application/json").expectContinue(true).POST(BodyPublishers.ofString(RequestBody)).build(),
+            r -> BodySubscribers.ofString(StandardCharsets.UTF_8)
+        );
+
+        assertEquals(response.statusCode(), 200);
+        assertEquals(response.body(), ExpectedResponse);
+        assertTrue(validated.get());
+      }
     }
   }
 
@@ -120,4 +126,4 @@ public void expectReject(String scheme) throws Exception {
       assertEquals(response.body(), "");
     }
   }
-}
\ No newline at end of file
+}

From 32746a9638ac0ca06305366d39744d45ba87560d Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Thu, 15 May 2025 23:14:38 -0600
Subject: [PATCH 12/15] Tests

---
 .../fusionauth/http/server/HTTPRequest.java   | 16 ++++---
 .../http/server/io/HTTPInputStream.java       |  8 +++-
 .../io/fusionauth/http/util/HTTPTools.java    |  3 +-
 .../java/io/fusionauth/http/CoreTest.java     | 45 +++++++++++++++++++
 4 files changed, 62 insertions(+), 10 deletions(-)

diff --git a/src/main/java/io/fusionauth/http/server/HTTPRequest.java b/src/main/java/io/fusionauth/http/server/HTTPRequest.java
index 0441be9..ace183c 100644
--- a/src/main/java/io/fusionauth/http/server/HTTPRequest.java
+++ b/src/main/java/io/fusionauth/http/server/HTTPRequest.java
@@ -264,14 +264,16 @@ public String getBaseURL() {
   }
 
   public byte[] getBodyBytes() throws BodyException {
-    if (bodyBytes == null && inputStream != null) {
-      try {
-        bodyBytes = inputStream.readAllBytes();
-      } catch (IOException e) {
-        throw new BodyException("Unable to read the HTTP request body bytes", e);
+    if (bodyBytes == null) {
+      if (inputStream != null) {
+        try {
+          bodyBytes = inputStream.readAllBytes();
+        } catch (IOException e) {
+          throw new BodyException("Unable to read the HTTP request body bytes", e);
+        }
+      } else {
+        bodyBytes = new byte[0];
       }
-    } else {
-      bodyBytes = new byte[0];
     }
 
     return bodyBytes;
diff --git a/src/main/java/io/fusionauth/http/server/io/HTTPInputStream.java b/src/main/java/io/fusionauth/http/server/io/HTTPInputStream.java
index bc37a2e..3f16cc9 100644
--- a/src/main/java/io/fusionauth/http/server/io/HTTPInputStream.java
+++ b/src/main/java/io/fusionauth/http/server/io/HTTPInputStream.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2024-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -89,6 +89,8 @@ public int read() throws IOException {
     }
 
     int b;
+    // If bodyBytes exist, they are left over bytes from parsing the preamble.
+    // - Process these bytes first before reading from the delegate InputStream.
     if (bodyBytes != null) {
       b = bodyBytes[bodyBytesIndex++];
 
@@ -124,6 +126,8 @@ public int read(byte[] buffer, int offset, int length) throws IOException {
     }
 
     int read;
+    // If bodyBytes exist, they are left over bytes from parsing the preamble.
+    // - Process these bytes first before reading from the delegate InputStream.
     if (bodyBytes != null) {
       read = Math.min(bodyBytes.length, length);
       System.arraycopy(bodyBytes, bodyBytesIndex, buffer, offset, read);
@@ -133,7 +137,7 @@ public int read(byte[] buffer, int offset, int length) throws IOException {
         bodyBytes = null;
       }
     } else {
-      read = delegate.read(buffer);
+      read = delegate.read(buffer, offset, length);
     }
 
     if (instrumenter != null) {
diff --git a/src/main/java/io/fusionauth/http/util/HTTPTools.java b/src/main/java/io/fusionauth/http/util/HTTPTools.java
index 72be250..6e6d474 100644
--- a/src/main/java/io/fusionauth/http/util/HTTPTools.java
+++ b/src/main/java/io/fusionauth/http/util/HTTPTools.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2023, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -298,6 +298,7 @@ public static byte[] parseRequestPreamble(InputStream inputStream, HTTPRequest r
     byte[] leftover = null;
     if (index < read) {
       logger.trace("Had [{}] body bytes from the preamble read.", (read - index));
+      // TODO : Daniel : Performance - can we just return an offset into the requestBuffer instead of copying these bytes?
       leftover = Arrays.copyOfRange(requestBuffer, index, read);
     }
 
diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index caa8b6f..e2dabfd 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -367,6 +367,51 @@ public void largeCSS(String scheme) throws Exception {
     }
   }
 
+  @Test
+  public void large_body() throws Exception {
+    // Ensure that when the body bytes overflow from the initial "left over" bytes read
+    // during reading of the preamble, the remaining bytes read from the HTTPInputStream
+    // properly use offset and lengths when reading.
+
+    var payload = "foo=" + "1234567890".repeat(12 * 1024);
+    byte[] bytes = payload.getBytes(StandardCharsets.UTF_8);
+
+    HTTPHandler handler = (req, res) -> {
+      res.setStatus(200);
+      res.setContentType("application/x-www-form-urlencoded");
+      res.setContentLength(req.getBodyBytes().length);
+      res.getOutputStream().write(req.getBodyBytes());
+      res.getOutputStream().close();
+
+      // This will hose up the works - or it did until we fixed it.
+      req.getFormData();
+    };
+
+    var instrumenter = new CountingInstrumenter();
+    try (var ignore = makeServer("http", handler, instrumenter).start(); var socket = new Socket("127.0.0.1", 4242)) {
+      var out = socket.getOutputStream();
+      var request = """
+          GET / HTTP/1.1\r
+          Host: localhost:42\r
+          Content-Type: application/x-www-form-urlencoded\r
+          Content-Length: {contentLength}\r
+          \r
+          {body}
+          """.replace("{contentLength}", "" + bytes.length)
+             .replace("{body}", payload);
+      out.write(request.getBytes());
+      out.flush();
+      var in = socket.getInputStream();
+
+      var responseBytes = in.readAllBytes();
+      var responseString = new String(responseBytes);
+
+      assertEquals(responseBytes.length, 122_999);
+      var payloadIndex = responseString.indexOf("\r\n\r\n") + 4;
+      assertEquals(responseString.substring(payloadIndex, responseBytes.length - 1), payload);
+    }
+  }
+
   @Test
   public void logger() {
     // Test replacement values and ensure we are handling special regex characters.

From 83561019ccdc02f307fb2cdf0f61aa85259a7885 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Fri, 16 May 2025 10:31:45 -0600
Subject: [PATCH 13/15] Tests

---
 build.savant                                  |  2 +-
 pom.xml                                       |  2 +-
 .../fusionauth/http/ClientAbortException.java | 31 -------------------
 .../server/internal/HTTPServerThread.java     |  4 ++-
 .../java/io/fusionauth/http/CoreTest.java     | 30 +++++++++---------
 5 files changed, 21 insertions(+), 48 deletions(-)
 delete mode 100644 src/main/java/io/fusionauth/http/ClientAbortException.java

diff --git a/build.savant b/build.savant
index 8350d52..c4c25f5 100644
--- a/build.savant
+++ b/build.savant
@@ -17,7 +17,7 @@ jackson5Version = "3.0.1"
 restifyVersion = "4.2.1"
 testngVersion = "7.10.2"
 
-project(group: "io.fusionauth", name: "java-http", version: "0.4.0-RC.6", licenses: ["ApacheV2_0"]) {
+project(group: "io.fusionauth", name: "java-http", version: "0.4.0-RC.7", licenses: ["ApacheV2_0"]) {
   workflow {
     fetch {
       // Dependency resolution order:
diff --git a/pom.xml b/pom.xml
index 00bb874..f7dd28e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,7 +2,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>io.fusionauth</groupId>
   <artifactId>java-http</artifactId>
-  <version>0.4.0-RC.6</version>
+  <version>0.4.0-RC.7</version>
   <packaging>jar</packaging>
 
   <name>Java HTTP library (client and server)</name>
diff --git a/src/main/java/io/fusionauth/http/ClientAbortException.java b/src/main/java/io/fusionauth/http/ClientAbortException.java
deleted file mode 100644
index 216edf6..0000000
--- a/src/main/java/io/fusionauth/http/ClientAbortException.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Copyright (c) 2023, FusionAuth, All Rights Reserved
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
- * either express or implied. See the License for the specific
- * language governing permissions and limitations under the License.
- */
-package io.fusionauth.http;
-
-import java.io.IOException;
-
-/**
- * An IOException that is most likely caused by the client closing a socket.
- * <p>
- * For example:
- * <p>
- * <code>java.io.IOException: Connection reset by peer</code>
- */
-public class ClientAbortException extends IOException {
-  public ClientAbortException(IOException e) {
-    super(e);
-  }
-}
diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
index 4e258c3..fd75ab3 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
@@ -232,7 +232,9 @@ public void run() {
 
             logger.debug("[{}] Closing connection readingSlow=[{}] writingSlow=[{}] timedOut=[{}] {}", threadId, readingSlow, writingSlow, timedOut, message);
             logger.debug("[{}] Closing client connection [{}] due to inactivity", threadId, worker.getSocket().getRemoteSocketAddress());
+          }
 
+          if (logger.isTraceEnabled()) {
             StringBuilder threadDump = new StringBuilder();
             for (Map.Entry<Thread, StackTraceElement[]> entry : Thread.getAllStackTraces().entrySet()) {
               threadDump.append(entry.getKey()).append(" ").append(entry.getKey().getState()).append("\n");
@@ -242,7 +244,7 @@ public void run() {
               threadDump.append("\n");
             }
 
-            logger.debug("Thread dump from server side.\n" + threadDump);
+            logger.trace("Thread dump from server side.\n" + threadDump);
           }
 
           try {
diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index e2dabfd..63d8927 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -35,6 +35,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.zip.InflaterInputStream;
 
 import com.inversoft.net.ssl.SSLTools;
@@ -830,6 +831,7 @@ public void simplePost(String scheme, int responseBufferSize) throws Exception {
   public void slowClient(String scheme) throws Exception {
     // Test a slow connection where the HTTP server is blocked because we cannot write to the output stream as fast as we'd like. The
     // default buffer on macOS seems to be 768k (from my testing). I set this to 8MB which should hopefully cause the writes to back up.
+    // - This tests the minimumWriteThroughput and writeThroughputCalculationDelayDuration
     byte[] bytes = new byte[1024 * 1024 * 8];
     new SecureRandom().nextBytes(bytes);
 
@@ -839,23 +841,23 @@ public void slowClient(String scheme) throws Exception {
       res.setContentLength(bytes.length);
       res.setStatus(200);
 
-      int interation = 1;
+      int iteration = 1;
       var out = res.getOutputStream();
-      for (int i = 0; i < bytes.length; i += 1024 * 16) {
-        out.write(bytes, i, 1024 * 16);
-        System.out.println("Wrote " + (interation++ * 16) + "k");
+      for (int i = 0; i < bytes.length; i += 1024 * 8) {
+        out.write(bytes, i, 1024 * 8);
+        System.out.println("> Wrote    [" + (8 * 1024) + "] bytes. Total bytes written  [" + (iteration++ * 8 * 1024) + "].");
       }
       out.close();
     };
 
-    // TODO : Daniel Review : What are we testing here? Is this testing the "reader too slow" which is the read throughput calculation?
-    //        Or the initial read timeout?
-    //        Neither were being changed from the defaults. Default bytes / second on the read is 16k.
-    //        The test comments below mention waiting for 2 seconds, so assuming that is referring to the initialReadTimeoutDuration?
-    //        This test is flaky, adding Connection: close on the client doesn't seem to help either.
-    //        It does seem to fail on http, but pass on https for some reason. Needs investigation.
+    // Set the min write throughput to 1 Megabit / second
+    // - In this case the server is trying to write the client, and it cannot do so fast enough.
     AtomicBoolean slept = new AtomicBoolean(false);
-    try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler).start()) {
+    AtomicInteger totalBytesReceived = new AtomicInteger();
+    try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler)
+        // By default, we will wait 5 seconds before we calculate write throughput, reduce to this to 1 ms for this test.
+        .withWriteThroughputCalculationDelayDuration(Duration.ofMillis(1))
+        .withMinimumWriteThroughput(1024* 1024 * 1024).start()) {
       URI uri = makeURI(scheme, "");
       client.send(
           // Don't keep this connection open because the timeouts aren't the same on a keep alive.
@@ -866,9 +868,9 @@ public void slowClient(String scheme) throws Exception {
               // Sleep once since the server should fail after the first batch, but since Java or the OS might cache a lot of bytes it
               // read from the socket, we can't sleep for too long, otherwise, this test will never complete
               if (!slept.get()) {
-                int sleep = 5 * 1_000;
-                System.out.println("Received [" + actual.length + "] bytes. Sleep [" + sleep + "]");
-                sleep(sleep); // We expect to only wait for 2,000 ms
+                int sleep = 5_000;
+                System.out.println("> Received [" + actual.length + "] bytes. Total bytes received [" + totalBytesReceived.addAndGet(actual.length) + "]. Sleep [" + sleep + "] ms.");
+                sleep(sleep); // We expect to only wait for 2,000 ms until the cleaner runs to find slow clients.
                 slept.set(true);
               }
             } else {

From af92fe8c02e9d408d8e6872de38b1959f0476c42 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Fri, 16 May 2025 15:22:32 -0600
Subject: [PATCH 14/15] Tests

---
 .../io/fusionauth/http/server/internal/HTTPServerThread.java | 2 +-
 src/test/java/io/fusionauth/http/CoreTest.java               | 5 ++++-
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
index fd75ab3..5bb1e8d 100644
--- a/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
+++ b/src/main/java/io/fusionauth/http/server/internal/HTTPServerThread.java
@@ -157,7 +157,7 @@ public HTTPServerCleanerThread() {
 
     public void run() {
       while (running) {
-        logger.trace("Wake up. Clean up server threads.");
+        logger.trace("Wake up. Clean up client worker threads.");
 
         Iterator<ClientInfo> iterator = clients.iterator();
         while (iterator.hasNext()) {
diff --git a/src/test/java/io/fusionauth/http/CoreTest.java b/src/test/java/io/fusionauth/http/CoreTest.java
index 63d8927..f3b4249 100644
--- a/src/test/java/io/fusionauth/http/CoreTest.java
+++ b/src/test/java/io/fusionauth/http/CoreTest.java
@@ -279,7 +279,7 @@ public void initialReadTimeout() {
     var instrumenter = new CountingInstrumenter();
     try (var ignore = makeServer("http", handler, instrumenter).withInitialReadTimeout(Duration.ofMillis(250)).start(); var socket = new Socket("127.0.0.1", 4242)) {
       var out = socket.getOutputStream();
-      sleep(2_000);
+      sleep(2_500);
       out.write("""
           GET / HTTP/1.1\r
           Host: localhost:42\r
@@ -386,6 +386,9 @@ public void large_body() throws Exception {
 
       // This will hose up the works - or it did until we fixed it.
       req.getFormData();
+
+      // Ensure you can call this method and get the same value in return each time.
+      assertEquals(req.getBodyBytes(), req.getBodyBytes());
     };
 
     var instrumenter = new CountingInstrumenter();

From 8ae5cf13072ddd8178812f917a78308c397bdd79 Mon Sep 17 00:00:00 2001
From: Daniel DeGroff <daniel@fusionauth.io>
Date: Fri, 16 May 2025 15:28:49 -0600
Subject: [PATCH 15/15] copyright

---
 src/test/java/io/fusionauth/http/io/MultipartStreamTest.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java b/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java
index 3326c06..b3dc526 100644
--- a/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java
+++ b/src/test/java/io/fusionauth/http/io/MultipartStreamTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2023, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.