tricu

streaming-v2.tri (9609B)

---

 !import base as B
 !import native_math as N
 !import io_v2 as IO
 !import io_v2_http as Transport
 !import io_v2_artifacts as Stream
 !import http as HTTP
 !import handlers as H
 !import handler_artifacts as Artifacts
 !import handler_log as Log
 !import arboricx_handlers as Registry
 !import arboricx_server as Store
 !import conversions as Convert
 !export pump
 !export connection
 !export serve
 !export serveCount
 
 pump _ 0 _ = H.pure (B.ok t t)
 pump client remaining initial = let
   limit = N.minimum remaining 4096
   consume (unused, (reversed, rest)) = case reversed of
     [] = H.bind (H.liftIO (IO.readSome client limit)) (reply : case reply of
       ((t t), (("data", chunk), t)) = pump client remaining chunk
       _ = H.pure (B.err "Unexpected end of request body" t))
     _ = H.bind (Stream.write (B.reverse reversed)) (reply : case reply of
       ((t t), (t, t)) = pump client (N.subtract remaining (N.subtract limit unused)) rest
       _ = H.pure reply)
   consume _ = H.fail "invalid-response" "artifact-stream" "buffer" "Invalid body state"
   in consume (HTTP.takeBodyBytes initial limit [])
 
 logged sink status response = H.bind
   (Log.withSink sink (Log.withContext [["method" "POST"] ["target" "/_arboricx/bundle"]] (Log.info status)))
   (reply : case reply of
     ((t t), (t, t)) = H.pure response
     _ = H.pure (HTTP.errorResponse 500 "Internal Server Error\n"))
 upload root sink client count initial = H.bind
   (Artifacts.publishStreamV2 root count (H.label "streaming-upload" (pump client count initial)))
   (reply : case reply of
     ((t t), (digest, t)) = logged sink 201 (HTTP.createdResponse digest)
     (t, (_, t)) = logged sink 400 (HTTP.badRequestResponse "Upload failed\n")
     _ = H.fail "invalid-response" "artifact-stream" "publish" "Invalid upload Result")
 -- Supervised request interpretation retains the socket in its connection owner;
 -- borrowing does not extend lifetime. Join before responding, so cleanup failure
 -- cannot falsely acknowledge success. Cancellation is never rendered as HTTP 500.
 response root sink client count initial = IO.scope (_ : IO.require
   (IO.spawn 1 1 [] (_ : H.runIOV2 (upload root sink client count initial))) (spawned : case spawned of
     [_ ticket] = IO.require (IO.join ticket) (outcome : case outcome of
       ["completed" value] = IO.pure value
       ["failed" _ _] = IO.pure (HTTP.errorResponse 500 "Internal Server Error\n")
       ["cancelled" reason _] = IO.cancelled reason
       _ = IO.fail "Invalid upload task outcome")
     _ = IO.fail "Invalid upload spawn reply"))
 boundedResponse root limit sink client count initial = case N.lte count limit of
   (t t) = IO.bind (response root sink client count initial) (Transport.respond client "POST")
   _ = Transport.respond client "POST" (HTTP.errorResponse 413 "Request body too large\n")
 framed root limit sink client fields initial = case HTTP.contentLength fields of
   ((t t), (t, _)) = boundedResponse root limit sink client 0 initial
   ((t t), ((t count), _)) = boundedResponse root limit sink client count initial
   (t, (status, message)) = Transport.respond client "POST" (HTTP.errorResponse status message)
   _ = Transport.respond client "POST" (HTTP.badRequestResponse "Bad Request\n")
 headers root limit sink client version raw = case HTTP.parseHeaders raw of
   ((t t), (fields, initial)) = case HTTP.validateRequest "POST" "/_arboricx/bundle" version fields of
     ((t t), _) = framed root limit sink client fields initial
     (t, (status, message)) = Transport.respond client "POST" (HTTP.errorResponse status message)
     _ = Transport.respond client "POST" (HTTP.badRequestResponse "Bad Request\n")
   _ = Transport.respond client "" (HTTP.badRequestResponse "Bad Request\n")
 -- Log before committing headers. Supervision contains logger failures without
 -- letting a post-header transfer failure generate a second HTTP response.
 downloadLog sink method target status = IO.scope (_ : IO.require
   (IO.spawn 1 1 [] (_ : H.runIOV2
     (Log.withSink sink (Log.withContext [["method" method] ["target" target]] (Log.info status)))))
   (spawned : case spawned of
     [_ ticket] = IO.require (IO.join ticket) (outcome : case outcome of
       ["completed" ((t t), (t, t))] = IO.pure (t t)
       ["cancelled" reason _] = IO.cancelled reason
       _ = IO.pure t)
     _ = IO.fail "Invalid download logger task"))
 downloadError sink client method target status message = IO.bind
   (downloadLog sink method target status) (logged : case logged of
     (t t) = Transport.respond client method (HTTP.errorResponse status message)
     _ = Transport.respond client method (HTTP.errorResponse 500 "Internal Server Error\n"))
 prepareDownload owner root limit digest = do IO.onOk
   reader <- IO.openReader owner (Store.bundleObjectPath root digest)
   snapshot <- IO.snapshotBytes reader (Store.pathJoin root "tmp") limit digest
   IO.close reader
   IO.pure (B.ok snapshot t)
 sendDownload sink client method target [reader size] = IO.bind
   (downloadLog sink method target 200) (logged : case logged of
     (t t) = do IO.bind
       Transport.respond client method (HTTP.buildResponse 200
         [(B.pair "Content-Type" Store.bundleContentType)
          (B.pair "Content-Length" (Convert.showNumber size))
          (B.pair "Connection" "close")] [])
       case method of
         "HEAD" = IO.pure t
         _ = IO.require (IO.transferBytes reader client size) (_ : IO.pure t)
     _ = Transport.respond client method (HTTP.errorResponse 500 "Internal Server Error\n"))
 sendDownload _ _ _ _ _ = IO.fail "Invalid snapshot reply"
 download root limit sink client method target digest = case B.checkContract Store.bundleHashC digest of
   ((t t), (checked, t)) = IO.scope (owner : IO.bind
     (prepareDownload owner root limit checked) (reply : case reply of
       ((t t), (snapshot, t)) = sendDownload sink client method target snapshot
       _ = downloadError sink client method target 404 "Artifact unavailable\n"))
   (t, (message, _)) = downloadError sink client method target 400 message
   _ = IO.fail "Invalid digest contract reply"
 downloadRoute root limit sink client method target fields bytes = case HTTP.routerMethod method of
   "GET" = case Store.stripPrefix Store.bundleHashPrefix target of
     (t digest) = download root limit sink client method target digest
     _ = IO.bind (Registry.routerV2 root limit sink "GET" target fields bytes) (Transport.respond client method)
   routed = IO.bind (Registry.routerV2 root limit sink routed target fields bytes) (Transport.respond client method)
 
 line root limit sink client ((t t), (("POST", ("/_arboricx/bundle", version)), raw)) =
   headers root limit sink client version raw
 line root limit sink client parsed = Transport.handleWith (downloadRoute root limit sink) client parsed
 connection root limit sink client = IO.bind (Transport.recvHeaders client) (reply : case reply of
   ((t t), (raw, _)) = line root limit sink client (HTTP.parseRequestLine raw)
   (t, (431, _)) = Transport.respond client "" (HTTP.errorResponse 431 "Request Header Fields Too Large\n")
   _ = Transport.respond client "" (HTTP.badRequestResponse "Bad Request\n"))
 
 worker root limit sink [client] = connection root limit sink client
 worker _ _ _ _ = IO.fail "Invalid connection handoff"
 settle ticket = IO.require (IO.join ticket) (outcome : case outcome of
   ["completed" _] = IO.pure t
   ["failed" _ _] = IO.bind (Registry.writeLogV2 "registry connection failed\n") (_ : IO.pure t)
   ["cancelled" reason _] = IO.cancelled reason
   _ = IO.fail "Invalid connection task outcome")
 drain [] = IO.pure t
 drain [ticket | rest] = IO.bind (settle ticket) (_ : drain rest)
 drain _ = IO.fail "Invalid connection tickets"
 next (t count) = B.just (N.predecessor count)
 next _ = t
 -- FIFO bounded admission: consume the oldest ticket before accepting beyond the
 -- configured window. No discarded tickets, detached tasks or unbounded backlog in
 -- the language. A slow oldest connection can delay admission; no deadline promise.
 admit _ _ _ _ _ _ (t 0) tickets = drain tickets
 admit root limit sink owner listener window remaining tickets = case N.gte (B.length tickets) window of
   (t t) = case tickets of
     [ticket | rest] = IO.bind (settle ticket) (_ : admit root limit sink owner listener window remaining rest)
     _ = IO.fail "Invalid admission window"
   _ = IO.require (IO.acceptTCP listener owner) (client : IO.require
     (IO.spawn 1 1 [client] (worker root limit sink)) (spawned : case spawned of
       [_ ticket] = admit root limit sink owner listener window (next remaining) (B.append tickets [ticket])
       _ = IO.fail "Invalid connection spawn reply"))
 start root limit sink address window remaining ready = case B.checkContract B.nat? limit of
   ((t t), _) = startWindow root limit sink address window remaining ready
   _ = IO.fail "Invalid artifact byte limit"
 startWindow root limit sink address window remaining ready = case B.checkContract B.nat? window of
   ((t t), _) = case B.and? (N.gte window 1) (N.lte window 256) of
     (t t) = IO.scope (owner : do IO.require
       H.runIOV2 (Artifacts.prepareFilesystemV2 root)
       listener <- IO.listenTCP owner address 128
       IO.bind (ready listener) (_ : admit root limit sink owner listener window remaining []))
     _ = IO.fail "Connection window must be between 1 and 256"
   _ = IO.fail "Invalid connection window"
 serve root limit sink address window ready = start root limit sink address window t ready
 serveCount root limit sink address window count ready = case B.checkContract B.nat? count of
   ((t t), _) = start root limit sink address window (B.just count) ready
   _ = IO.fail "Invalid connection count"