fix: ext chunk

Author: algebraic-dev

src/Std/Internal/Http/Protocol/H1.lean

@@ -247,7 +247,6 @@ private def resetForNextMessage (machine : Machine ty) : Machine ty :=
         knownSize := none,
         messageHead := {},
         userClosedBody := false,
-        canSendData := true,
         sentMessage := false
       },
       events := machine.events.push .next,
@@ -347,7 +346,7 @@ def closeReader (machine : Machine dir) : Machine dir :=
 /-- Signal that the writer cannot send more messages because the socket closed. -/
 @[inline]
 def closeWriter (machine : Machine dir) : Machine dir :=
-  machine.modifyWriter ({ · with canSendData := false })
+  machine.modifyWriter ({ · with state := .closed, userClosedBody := true })
 
 /-- Signal that the user is not sending data anymore. -/
 @[inline]
@@ -416,16 +415,9 @@ def startNextCycle (machine : Machine dir) : Machine dir :=
 partial def processWrite (machine : Machine dir) : Machine dir :=
   match machine.writer.state with
   | .pending =>
-      if ¬machine.writer.canSendData ∨ machine.isReaderClosed then
-        machine.setWriterState .closed
-      else
-        machine
+      machine
   | .waitingHeaders =>
-      if ¬machine.writer.canSendData then
-        machine.setWriterState .closed
-      else
-        machine.addEvent .needAnswer
-
+      machine.addEvent .needAnswer
   | .waitingForFlush =>
       if machine.shouldFlush then
         machine.setHeaders machine.writer.messageHead
@@ -553,25 +545,24 @@ partial def processRead (machine : Machine dir) : Machine dir :=
       match result with
       | some (size, ext) =>
           machine
-          |>.setReaderState (.needChunkedBody size)
-          |>.setEvent (some ext <&> .chunkExt)
+          |>.setReaderState (.needChunkedBody ext size)
           |> processRead
       | none =>
           machine
 
-  | .needChunkedBody 0 =>
+  | .needChunkedBody ext 0 =>
     let (machine, result) := parseWith machine (parseLastChunkBody machine.config) (limit := some 1)
 
     match result with
     | some _ =>
         machine
         |>.setReaderState .complete
-        |>.addEvent (.gotData true .empty)
+        |>.addEvent (.gotData true ext .empty)
         |> processRead
     | none =>
         machine
 
-  | .needChunkedBody size =>
+  | .needChunkedBody ext size =>
       let (machine, result) := parseWith machine
         (parseChunkedSizedData size) (limit := none) (some size)
 
@@ -580,19 +571,19 @@ partial def processRead (machine : Machine dir) : Machine dir :=
         | .complete body =>
             machine
             |>.setReaderState .needChunkedSize
-            |>.addEvent (.gotData false body)
+            |>.addEvent (.gotData false ext body)
             |> processRead
         | .incomplete body remaining =>
             machine
-            |>.setReaderState (.needChunkedBody remaining)
-            |>.addEvent (.gotData false body)
+            |>.setReaderState (.needChunkedBody ext remaining)
+            |>.addEvent (.gotData false ext body)
       else
         machine
 
   | .needFixedBody 0 =>
       machine
       |>.setReaderState .complete
-      |>.addEvent (.gotData true .empty)
+      |>.addEvent (.gotData true #[] .empty)
       |> processRead
 
   | .needFixedBody size =>
@@ -603,20 +594,17 @@ partial def processRead (machine : Machine dir) : Machine dir :=
         | .complete body =>
             machine
             |>.setReaderState .complete
-            |>.addEvent (.gotData true body)
+            |>.addEvent (.gotData true #[] body)
             |> processRead
         | .incomplete body remaining =>
             machine
             |>.setReaderState (.needFixedBody remaining)
-            |>.addEvent (.gotData false body)
+            |>.addEvent (.gotData false #[] body)
       else
         machine
 
-  | .requireOutgoing _ =>
-      machine
-
   | .complete =>
-      if ¬machine.keepAlive then
+      if ¬machine.keepAlive ∨ machine.reader.noMoreInput then
         machine.setReaderState .closed
       else
         machine

src/Std/Internal/Http/Protocol/H1/Event.lean

@@ -22,11 +22,6 @@ namespace Std.Http.Protocol.H1
 Events that can occur during HTTP message processing.
 -/
 inductive Event (dir : Direction)
-  /--
-  Event received when chunk extension data is encountered in chunked encoding.
-  -/
-  | chunkExt (ext : Array (String × Option String))
-
   /--
   Event received when the headers end.
   -/
@@ -35,7 +30,7 @@ inductive Event (dir : Direction)
   /--
   Event received when some data arrives from the received message.
   -/
-  | gotData (final : Bool) (data : ByteSlice)
+  | gotData (final : Bool) (ext : Array (String × Option String)) (data : ByteSlice)
 
   /--
   Event received when the input ends and more data is required to continue

src/Std/Internal/Http/Protocol/H1/Parser.lean

@@ -78,8 +78,9 @@ def token (limit : Nat) : Parser ByteSlice :=
   takeWhileUpTo1 isTokenCharacter limit
 
 @[inline]
-def crlf : Parser Unit :=
-  skipBytes "\r\n".toUTF8
+def crlf : Parser Unit := do
+  discard <| optional (skipByte '\r'.toUInt8)
+  skipByte '\n'.toUInt8
 
 @[inline]
 def rsp (limits : H1.Config) : Parser Unit :=
@@ -118,9 +119,16 @@ def parseHttpVersion : Parser Version := do
   opt <| Version.fromNumber? (major - 48 |>.toNat) (minor - 48 |>.toNat)
 
 --   method         = token
-def parseMethod (limits : H1.Config) : Parser Method := do
-  let method ← token limits.maxMethodLength
-  opt <| Method.fromString? =<< (String.fromUTF8? method.toByteArray)
+def parseMethod : Parser Method :=
+  (skipBytes "GET".toUTF8 <&> fun _ => Method.get)
+  <|> (skipBytes "HEAD".toUTF8 <&> fun _ => Method.head)
+  <|> (attempt <| skipBytes "POST".toUTF8 <&> fun _ => Method.post)
+  <|> (skipBytes "PUT".toUTF8 <&> fun _ => Method.put)
+  <|> (skipBytes "DELETE".toUTF8 <&> fun _ => Method.delete)
+  <|> (skipBytes "CONNECT".toUTF8 <&> fun _ => Method.connect)
+  <|> (skipBytes "OPTIONS".toUTF8 <&> fun _ => Method.options)
+  <|> (skipBytes "TRACE".toUTF8 <&> fun _ => Method.trace)
+  <|> (skipBytes "PATCH".toUTF8 <&> fun _ => Method.patch)
 
 def parseURI (limits : H1.Config) : Parser ByteArray := do
   let uri ← takeUntilUpTo (· == ' '.toUInt8) limits.maxUriLength
@@ -132,7 +140,7 @@ Parses a request line
 request-line   = method SP request-target SP HTTP-version
 -/
 public def parseRequestLine (limits : H1.Config) : Parser Request.Head := do
-  let method ← parseMethod limits <* rsp limits
+  let method ← parseMethod <* rsp limits
   let uri ← parseURI limits <* rsp limits
 
   let uri ← match (Std.Http.Parser.parseRequestTarget <* eof).run uri with
@@ -157,8 +165,12 @@ Parses a single header.
 
 field-line CRLF / CRLF
 -/
-public def parseSingleHeader (limits : H1.Config) : Parser (Option (String × String)) :=
-  optional (attempt <| parseFieldLine limits) <* crlf
+public def parseSingleHeader (limits : H1.Config) : Parser (Option (String × String)) := do
+  if (← peek?) == some '\r'.toUInt8 ∨ (← peek?) == some '\n'.toUInt8 then
+    crlf
+    pure none
+  else
+    some <$> (parseFieldLine limits <* crlf)
 
 -- quoted-pair = "\" ( HTAB / SP / VCHAR / obs-text )
 def parseQuotedPair : Parser UInt8 := do

src/Std/Internal/Http/Protocol/H1/Reader.lean

@@ -37,18 +37,13 @@ inductive Reader.State (dir : Direction) : Type
   /--
   State waiting for chunk body data of specified size.
   -/
-  | needChunkedBody : Nat → State dir
+  | needChunkedBody : Array (String × Option String) → Nat → State dir
 
   /--
   State waiting for fixed-length body data of specified size.
   -/
   | needFixedBody : Nat → State dir
 
-  /--
-  Requires the outgoing message to continue.
-  -/
-  | requireOutgoing : Body.Length → State dir
-
   /--
   State that it completed a single request and can go to the next one
   -/

src/Std/Internal/Http/Protocol/H1/Writer.lean

@@ -103,11 +103,6 @@ structure Writer (dir : Direction) where
   -/
   userClosedBody : Bool := false
 
-  /--
-  Can send data to the output, it's used when the connection is closed and cannot send more data.
-  -/
-  canSendData : Bool := true
-
 namespace Writer
 
 /--

src/Std/Internal/Http/Server/Config.lean

@@ -44,12 +44,12 @@ structure Config where
   /--
   Timeout for keep-alive connections
   -/
-  keepAliveTimeout : { x : Time.Millisecond.Offset // 0 < x } :=  ⟨45000, by decide⟩
+  keepAliveTimeout : { x : Time.Millisecond.Offset // 0 < x } :=  ⟨12000, by decide⟩
 
   /--
   Maximum timeout time for request more data.
   -/
-  requestTimeout : { x : Time.Millisecond.Offset // 0 < x } := ⟨10000, by decide⟩
+  requestTimeout : { x : Time.Millisecond.Offset // 0 < x } := ⟨13000, by decide⟩
 
   /--
   Whether to enable keep-alive connections by default.

src/Std/Internal/Http/Server/Connection.lean

@@ -131,9 +131,6 @@ private def handle
     if step.output.size > 0 then
       Transport.sendAll socket step.output.data
 
-    if machine.reader.state == .closed ∧ ¬waitingResponse then
-      machine := machine.closeWriter
-
     for event in step.events do
       match event with
       | .needMoreData expect => do
@@ -157,8 +154,8 @@ private def handle
 
         BaseIO.chainTask task fun x => discard <| response.resolve x
 
-      | .gotData final data =>
-        discard <| requestStream.write data.toByteArray
+      | .gotData final ext data =>
+        requestStream.writeChunk { data := data.toByteArray, extensions := ext }
 
         if final then
           requestStream.close
@@ -170,13 +167,10 @@ private def handle
         respStream := none
 
       | .failed _ =>
-        do pure ()
+        pure ()
 
       | .close =>
-        do pure ()
-
-      | .chunkExt _=>
-        do pure ()
+        pure ()
 
     if requiresData ∨ needAnswer ∨ respStream.isSome then
       let socket := if requiresData then some socket else none
@@ -204,18 +198,24 @@ private def handle
         machine := machine.closeReader
         if machine.isWaitingMessage ∧ waitingResponse then
           waitingResponse := false
-          machine := machine.send { status := .requestTimeout }
+          machine := machine.send { status := .requestTimeout, headers := .empty |>.insert (.new "connection") (.new "close") }
           machine := machine.userClosedBody
+          machine := machine.noMoreInput
           respStream := none
         else
           machine := machine.closeWriter
+          |>.noMoreInput
 
       | .response (.error _) =>
         if machine.isWaitingMessage ∧ waitingResponse then
           waitingResponse := false
-          machine := machine.send { status := .internalServerError }
+          machine := machine.send { status := .internalServerError, headers := .empty |>.insert (.new "connection") (.new "close") }
           machine := machine.userClosedBody
           machine := machine.closeReader
+          machine := machine.noMoreInput
+        else
+          machine := machine.closeWriter
+          |>.noMoreInput
 
       | .response (.ok res) =>
         machine := machine.send res.head