Brotli compression

.github/workflows/lean_action_ci.yml

@@ -16,6 +16,28 @@ jobs:
       - name: Lint the library
         run: lake lint --builtin-only --linters=.missingDocs Datastar
 
+  # The Brotli codec is a separate package with C sources, built by Lake with the system compiler.
+  brotli:
+    runs-on: ubuntu-latest
+    needs: build
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Build the library
+        uses: leanprover/lean-action@v1
+        with:
+          build-args: "Datastar"
+
+      - name: Test datastar-brotli
+        run: lake -d datastar-brotli test
+
+      - name: Lint datastar-brotli
+        run: lake -d datastar-brotli lint --builtin-only --linters=.missingDocs DatastarBrotli
+
+      - name: Test datastar-brotli under sanitizers
+        run: datastar-brotli/scripts/sanitize.sh
+
   e2e:
     runs-on: ubuntu-latest
     needs: build
@@ -29,6 +51,10 @@ jobs:
         with:
           build-args: "e2e-server"
 
+      # Built here so that Playwright does not time out waiting for it to compile.
+      - name: Build brotli-e2e-server
+        run: lake -d datastar-brotli build brotli-e2e-server
+
       - uses: actions/setup-node@v4
         with:
           node-version: "20"

.gitignore

@@ -1,5 +1,7 @@
 /.lake
 
+.vscode/c_cpp_properties.json
+
 tmp/
 
 node_modules/

.vscode/c_cpp_properties.sample.json

@@ -0,0 +1,16 @@
+// Copy to c_cpp_properties.json (gitignored) and set the toolchain include
+// path to match lean-toolchain; `lean --print-prefix` prints the prefix.
+{
+  "version": 4,
+  "configurations": [
+    {
+      "name": "datastar-brotli",
+      "includePath": [
+        "${env:HOME}/.elan/toolchains/leanprover--lean4---v4.34.1/include",
+        "${workspaceFolder}/datastar-brotli/vendor/brotli/brotli-1.2.0/c/include"
+      ],
+      "defines": ["NDEBUG", "DATASTAR_BROTLI_TESTING"],
+      "cStandard": "c11"
+    }
+  ]
+}

Datastar/StdHttp.lean

@@ -110,8 +110,23 @@ private def sseResponseCore
     let sse : ServerSentEventGenerator := {stream, lock, encoder, finished}
     try
       ContextAsync.runIn ctx (callback sse)
-    finally
+    catch e =>
+      -- The callback threw, or the client disconnected. We should try to finish the
+      -- sse stream, as this will clean up resources like the encoder.
+      -- This call to `sse.finish` might also fail, but we drop the error here as we
+      -- are about to throw `e`.
       try sse.finish catch _ => pure ()
+      throw e
+
+    -- The callback completed, so finish the SSE stream, which calls `encoder.finish` for
+    -- a compressed stream.
+    --
+    -- We deliberately do not wrap this `sse.finish` in a try/catch. If we did so, an
+    -- error from `encoder.finish` would be squashed, and this (body) producer would
+    -- return normally. The client would receive a complete response whose compressed
+    -- stream is missing its final bytes. Instead, the error is propagated so it
+    -- reaches `Handler.onFailure` and the connection is closed as incomplete.
+    sse.finish
 
 /--
 A response that streams SSE events. The connection stays open until `callback` returns.

README.md

@@ -46,7 +46,7 @@ Add the package to `lakefile.toml`:
 ```toml
 [[require]]
 name = "datastar"
-git = "https://github.com/carlohamalainen/datastar-lean"
+git = "https://github.com/starfederation/datastar-lean"
 rev = "v0.1.0"
 ```
 
@@ -85,7 +85,10 @@ sseResponseWith [brotli, gzip] req fun sse =>
 
 If the client accepts none of them, the stream is sent uncompressed.
 
-**No codecs ship with this package yet.**
+No codecs ship with this package. Brotli is in [datastar-brotli](datastar-brotli/), a separate
+package in this repository.
+
+Codecs for zstd and zlib are in the pipeline.
 
 A codec is a `Compressor`, see [Datastar/Compression.lean](Datastar/Compression.lean).
 
@@ -104,6 +107,14 @@ Then open <http://127.0.0.1:3000>. Each example takes the port as an optional ar
 | `hello-world-channel` | State shared between connections; Start restarts the animation on every open page. |
 | `activity-feed` | `patchSignals` with `patchElements`; `@post` requests with signals in the body. |
 
+## Tests
+
+```
+lake test
+```
+
+checks the `#guard`s in `Test/Unit.lean` and runs `Test/StdHttp.lean`.
+
 ## SDK tests
 
 The official [Datastar SDK test suite](https://github.com/starfederation/datastar/tree/main/sdk/tests)

Test/Main.lean

@@ -0,0 +1,14 @@
+import Test.Unit
+import Test.StdHttp
+
+/-- Run by `lake test`. The `#guard`s of `Test.Unit` are checked as it compiles. -/
+def main : IO UInt32 := do
+  let mut failures := 0
+  for (name, test) in stdHttpTests do
+    try
+      test
+      IO.println s!"datastar: {name}: ok"
+    catch e =>
+      IO.eprintln s!"FAILED {name}: {e}"
+      failures := failures + 1
+  return if failures == 0 then 0 else 1

Test/StdHttp.lean

@@ -0,0 +1,76 @@
+import Datastar
+
+/-!
+The SSE response over a mock connection, run by `lake test`.
+
+Checked:
+
+* When the callback returns and `finish` fails, the response fails: the connection is dropped
+  without the chunked terminator, and `onFailure` sees the error. Ended normally, a compressed
+  stream would reach the client cut short, and nothing on the server would show it.
+* When `finish` succeeds, its bytes are sent and the response ends normally.
+* When the callback fails, that is the error reported, and a failing `finish` is then ignored.
+-/
+
+open Std Async Http Server
+open Datastar
+
+/-- A compressor that passes bytes through, ending the stream with `finish`. -/
+private def passthrough (finish : IO ByteArray) : Compressor where
+  encoding := "test"
+  start := pure { compress := pure, finish }
+
+private def finishFailure : IO ByteArray := throw (IO.userError "finish failed")
+
+private def callbackFailure : ServerSentEventGenerator → ContextAsync Unit :=
+  fun _ => throw (IO.userError "callback failed")
+
+private def oneEvent (sse : ServerSentEventGenerator) : ContextAsync Unit :=
+  sse.send <| patchSignals "{\"n\":1}"
+
+private def request : String :=
+  "GET /sse HTTP/1.1\r\nHost: example.com\r\nAccept-Encoding: test\r\nConnection: close\r\n\r\n"
+
+/-- The chunked terminator, which ends a response that completed. -/
+private def chunkEnd : String := "0\r\n\r\n"
+
+/-- Serve `request` to an SSE handler: the bytes written to the client, and the errors reported. -/
+private def serve (compressor : Compressor) (callback : ServerSentEventGenerator → ContextAsync Unit) :
+    IO (String × List String) := do
+  let failures ← IO.mkRef ([] : List String)
+  let handler := Handler.ofFns (fun req => sseResponseWith [compressor] req callback)
+    (onFailure := fun e => failures.modify (· ++ [toString e]))
+  let (client, server) ← Internal.Mock.new
+  let output ← Async.block do
+    client.send request.toUTF8
+    (serveConnection server handler { lingeringTimeout := 1000, generateDate := false }).run
+    client.recv?
+  return (String.fromUTF8! (output.getD .empty), ← failures.get)
+
+private def check (condition : Bool) (message : String) : IO Unit :=
+  unless condition do throw <| IO.userError message
+
+private def has (text needle : String) : Bool := (text.splitOn needle).length > 1
+
+private def finishFails : IO Unit := do
+  let (output, failures) ← serve (passthrough finishFailure) oneEvent
+  check (output.startsWith "HTTP/1.1 200") s!"response: {output.quote}"
+  check (has output "data: signals {\"n\":1}") s!"event not sent: {output.quote}"
+  check (!output.endsWith chunkEnd) s!"response ended as complete: {output.quote}"
+  check (failures == ["finish failed"]) s!"failures reported: {failures}"
+
+private def finishSucceeds : IO Unit := do
+  let (output, failures) ← serve (passthrough (pure "tail".toUTF8)) oneEvent
+  check (has output "data: signals {\"n\":1}") s!"event not sent: {output.quote}"
+  check (output.endsWith s!"tail\r\n{chunkEnd}") s!"final bytes or terminator missing: {output.quote}"
+  check (failures == []) s!"failures reported: {failures}"
+
+private def callbackFails : IO Unit := do
+  let (_, failures) ← serve (passthrough finishFailure) callbackFailure
+  check (failures == ["callback failed"]) s!"failures reported: {failures}"
+
+/-- The tests of this module, by name. -/
+def stdHttpTests : List (String × IO Unit) := [
+  ("finish fails after the callback returns", finishFails),
+  ("finish succeeds", finishSucceeds),
+  ("the callback fails", callbackFails)]

datastar-brotli/.gitignore

@@ -0,0 +1 @@
+/.lake

datastar-brotli/BrotliTest/E2EServer.lean

@@ -0,0 +1,57 @@
+import Datastar
+import DatastarBrotli
+
+import Std.Async
+import Std.Http
+import Std.Net.Addr
+
+/-!
+Serves the page that the Playwright test in `e2e/tests/brotli.spec.ts` drives.
+
+`/sse/stream` is a Brotli-compressed stream that stops after each event until the test calls
+`/release`. The test can then see that the browser has an event while the stream is still open,
+which is only true if every event is flushed through the encoder.
+-/
+
+open Std Async Http Server
+open Datastar
+
+def testPage : String := include_str "e2e-server.html"
+
+/-- A list of `count` items; 3000 of them are well over the encoder's 64 KiB input block. -/
+def largeList (count : Nat) : String :=
+  let items := (List.range count).map fun i => s!"<li id=\"item-{i}\">Item number {i}</li>"
+  s!"<ul id=\"large\">{String.join items}</ul>"
+
+/-- Wait until `/release` has been called `count` times. -/
+partial def waitForRelease (released : IO.Ref Nat) (count : Nat) : Async Unit := do
+  if (← released.get) < count then
+    sleep (.ofNat 20)
+    waitForRelease released count
+
+def app (released : IO.Ref Nat) (req : Request Body.Stream) : ContextAsync (Response Body.Any) := do
+  match req.line.method, toString req.line.uri.path with
+  | .get, "/" => Response.ok |>.html testPage
+  | .get, "/release" =>
+    released.modify (· + 1)
+    Response.ok |>.text "released"
+  | .get, "/sse/stream" =>
+    let start ← released.get
+    sseResponseWith [brotli] req fun sse => do
+      sse.send <| patchElements "<div id=\"step\">Event 1</div>"
+      waitForRelease released (start + 1)
+      sse.send <| patchElements (largeList 3000)
+      sse.send <| patchElements "<div id=\"step\">Event 2</div>"
+      waitForRelease released (start + 2)
+      sse.send <| patchElements "<div id=\"step\">Event 3</div>"
+  | _, _ => Response.notFound |>.text "Not found"
+
+def main : IO Unit := Async.block do
+  let addr : Std.Net.SocketAddressV4 := ⟨.ofParts 127 0 0 1, 3114⟩
+  let released ← IO.mkRef 0
+
+  let server ← Server.serve addr <| Handler.ofFn (app released)
+
+  IO.println "brotli-e2e-server running on http://127.0.0.1:3114"
+  (← IO.getStdout).flush
+  server.waitShutdown