commit 5c24105993a90f8579c7af6c64ed05cc1d832eb0
parent 8501bcd20465d6f883660fe49239006875a7e945
Author: James Eversole <james@eversole.co>
Date: Fri, 23 Oct 2026 20:41:06 -0500
Stream authenticated registry downloads through native byte snapshots
Add scoped SHA-256 snapshots and exact native byte transfers, keeping bundle bodies out of resident tree lists. Authenticate before headers and retain corruption, HEAD, logging, cancellation and disconnect controls.
Validate with nix flake check (921 native tests), arboricxServerTest and 54 HTTP controls. The 917828-byte site compiler downloads in 16-19 ms on localhost with about 21 MiB peak RSS. Add compiler latency and download retention regression gates; deployment remains separate.
Diffstat:
18 files changed, 541 insertions(+), 27 deletions(-)
diff --git a/docs/build-and-run.md b/docs/build-and-run.md
@@ -130,7 +130,11 @@ back up `registry-store` across replacements; do not mount over `/nix/store`.
The default runner uses IO v2 with OS process authority. Each connection owns its
socket and upload resources; admission waits for the oldest child at a window of
16 connections. A stalled oldest connection can delay later requests, including
-health checks. Uploads process bounded chunks; downloads retain whole values.
+health checks. Uploads process bounded chunks. Bundle downloads authenticate into
+anonymous disk snapshots and transfer native byte chunks; package metadata still
+uses whole-value reads. Each active snapshot uses at most the configured artifact
+byte limit in temporary disk space under `store/tmp`. Provision that filesystem
+for the admission window, including slow clients holding verified snapshots.
Storage initialization and socket bind/listen failures exit nonzero. Monitor the
HTTP health endpoint externally; the minimal image has no curl-based healthcheck.
@@ -158,8 +162,10 @@ actual runner: startup failures, health, binary uploads/downloads, invalid paths
package lookup, and persistence across restart. It also checks failed logging after
publication, fragmented uploads, early EOF and cleanup, FIFO admission, and repeated
64-KiB/1-MiB upload/download workloads. Reports under `retention/` record allocation,
-GC-sampled live heap and RSS; upload guards remain128MiB RSS and16MiB live heap.
-Large downloads retain substantial whole-value cost. The checks run the native
+GC-sampled live heap and RSS; upload and bundle-download guards are 128 MiB RSS
+and 16 MiB live heap. `compiler-download.json` measures first-byte and total
+latency for the actual site compiler, including concurrent requests, corruption,
+HEAD and client disconnects. The checks run the native
entrypoint directly. Container-engine integration requires a separate check.
## Explicit compiler selection
@@ -216,4 +222,4 @@ early-EOF cleanup, FIFO admission, and failed-logging-after-publication controls
Paths resolve under OS process authority, including symlink traversal. An object
can remain published after logging fails. See
[body retention measurements](validation.md.html#body-ingestion-and-retention-probes)
-for bounded upload checks and whole-value download limits.
+for upload/download regression guards and measurement limits.
diff --git a/docs/language-features.md b/docs/language-features.md
@@ -237,7 +237,9 @@ Evidence: [do AST lowering](../lib/source/ast.tri),
[scoped handlers](contracts.md.html#10-scoped-request-handlers).
`handler_upload` selects the scoped v2 producer/staged-publication composition;
see the [registry runner](build-and-run.md.html#deploy-the-arboricx-registry-linux).
- Downloads retain whole values; producer uploads consume bounded chunks.
+ Bundle downloads use authenticated native snapshots and bounded transfers;
+ producer uploads consume bounded chunks. Whole-value artifact APIs remain
+ available for callers needing resident byte lists.
- Contract helpers include `checkContract`, `withContract`, `guardC`,
collection/higher-order combinators, and `x@contract` / `=@contract` annotations.
Contracts and Boolean predicates have different protocols. Annotated success
diff --git a/docs/validation.md b/docs/validation.md
@@ -1451,8 +1451,8 @@ runtime compatibility. See [the registry runner](build-and-run.md.html#deploy-th
### Body ingestion and retention probes
-Measure the supported v2 server's source-owned staged uploads and whole-value
-downloads. The old discard/count/hash and whole-body v1 comparison runners are
+Measure the supported v2 server's source-owned staged uploads and authenticated
+native bundle downloads. The old discard/count/hash and whole-body v1 comparison runners are
retired; their historical measurements are not relabeled as v2 results.
```sh
@@ -1482,12 +1482,18 @@ concurrent maximum-size capacity, forced-GC behavior, or graceful shutdown.
The default phases are `streaming-publish` and `streaming-download`.
`--phases streaming-publish` selects uploads only; `--bounded-streaming` requires
-the upload phase's observed peak RSS to remain within128MiB and its GC-sampled
-maximum live heap within16MiB. Missing RTS statistics fail that guard rather than
-silently passing. Downloads retain whole values and are not covered by those
-upload limits. These are workload regression guards, not service guarantees.
+each selected streaming phase's observed peak RSS to remain within 128 MiB and
+its GC-sampled maximum live heap within 16 MiB. Missing RTS statistics fail that
+guard. These are workload regression guards, not service guarantees. Downloads
+use an anonymous authenticated disk snapshot bounded by the configured artifact
+limit per active request; native transfer avoids resident byte-list bodies.
`nix build .#arboricxServerTest --no-link` runs both phases at64KiB and1MiB with
three repetitions, alongside actual HTTP/security/lifecycle controls. This is
-end-to-end chunk consumption/publication, not a change of socket receive size
-alone.
+end-to-end chunk consumption/publication and authenticated native transfer.
+The same gate runs `test/registry_download.py` against the site compiler fixture,
+recording first-byte/total latency, process RSS and RTS metrics in
+`compiler-download.json`. It checks repeated and four-way concurrent GETs,
+authenticated HEAD, corruption, missing objects and recovery after a peer reset.
+A five-second localhost request ceiling catches timeout-scale regressions; it is
+not a production latency guarantee.
diff --git a/flake.nix b/flake.nix
@@ -107,6 +107,9 @@
--report $out/report.json
python3 ${./test/streaming_protocol.py} ${runner}/bin/${name} \
--io-version 2 > $out/protocol.txt
+ PYTHONDONTWRITEBYTECODE=1 python3 ${./.}/test/registry_download.py \
+ ${runner}/bin/${name} ${./test/fixtures/compiler-library.arboricx} \
+ --report $out/compiler-download.json > $out/compiler-download-summary.json
PYTHONDONTWRITEBYTECODE=1 python3 ${./.}/test/body_retention.py \
--runner ${runner}/bin/${name} \
--phases streaming-publish streaming-download --bounded-streaming \
diff --git a/lib/arboricx/streaming-v2.tri b/lib/arboricx/streaming-v2.tri
@@ -8,6 +8,8 @@
!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
@@ -62,9 +64,54 @@ headers root limit sink client version raw = case HTTP.parseHeaders raw of
(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.handle (Registry.routerV2 root limit sink) client parsed
+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")
diff --git a/lib/io-v2.tri b/lib/io-v2.tri
@@ -93,6 +93,15 @@ decodeUTF8 final bytes = request "utf8" 1 "decode" [final bytes]
readSome reader limit = request "stream" 1 "read" (B.pair reader limit)
writeSome writer bytes = request "stream" 1 "write" (B.pair writer bytes)
+-- Whole-stream SHA-256 authentication into an anonymous read-only file.
+-- Result: [reader size]. Scope-owned; disk <= limit, native chunks <=4096.
+-- An EOF probe may consume one byte beyond the limit on failure.
+snapshotBytes reader directory limit digest = request "stream" 1 "snapshot"
+ [reader directory limit digest]
+-- Exact native transfer from the current cursor. Failure can follow partial
+-- writes; callers close the destination rather than retrying uncertain progress.
+transferBytes reader writer count = request "stream" 1 "transfer" [reader writer count]
+
close handle = request "resource" 1 "close" handle
openStage owner directory = request "stage" 1 "open" (B.pair owner directory)
diff --git a/lib/io-v2/http-transport.tri b/lib/io-v2/http-transport.tri
@@ -6,6 +6,7 @@
!export readBody
!export respond
!export handle
+!export handleWith
-- Reuse pure HTTP grammar/limits. The header cap covers the entire accumulated
-- receive buffer, including prefetched body bytes, exactly as in the v1 reader.
@@ -39,8 +40,8 @@ send client bytes = case HTTP.takeBodyBytes bytes 4096 [] of
(_, (reversed, rest)) = IO.require (IO.writeAll client (B.reverse reversed)) (_ : send client rest)
_ = IO.fail "Invalid response byte list"
respond client method response = send client (HTTP.responseForMethod method response)
-body router client method target fields ((t t), (bytes, _)) = IO.bind
- (router (HTTP.routerMethod method) target fields bytes) (respond client method)
+body router client method target fields ((t t), (bytes, _)) =
+ router client method target fields bytes
body _ client method _ _ (t, (status, message)) = respond client method (HTTP.errorResponse status message)
body _ client method _ _ _ = respond client method (HTTP.badRequestResponse "Bad Request\n")
fields router client method target version raw = case HTTP.parseHeaders raw of
@@ -49,5 +50,12 @@ fields router client method target version raw = case HTTP.parseHeaders raw of
(t, (status, message)) = respond client method (HTTP.errorResponse status message)
_ = respond client method (HTTP.badRequestResponse "Bad Request\n")
_ = respond client "" (HTTP.badRequestResponse "Bad Request\n")
-handle router client ((t t), ((method, (target, version)), raw)) = fields router client method target version raw
-handle _ client _ = respond client "" (HTTP.badRequestResponse "Bad Request\n")
+handleWith router client ((t t), ((method, (target, version)), raw)) = fields router client method target version raw
+handleWith _ client _ = respond client "" (HTTP.badRequestResponse "Bad Request\n")
+
+-- Response callbacks run only after the same request grammar/body checks. They
+-- may send a bounded native body; framing errors use the ordinary byte response.
+handle router client parsed = handleWith
+ (socket method target headers bytes : IO.bind
+ (router (HTTP.routerMethod method) target headers bytes) (respond socket method))
+ client parsed
diff --git a/src/IOV2/ByteStream.hs b/src/IOV2/ByteStream.hs
@@ -0,0 +1,84 @@
+{-# LANGUAGE BangPatterns #-}
+-- Native bulk bytes; source owns paths, digest selection and response policy.
+module IOV2.ByteStream (snapshot, snapshotWith, transfer) where
+
+import qualified IOV2.Resources as R
+import Control.Exception
+import Control.Monad (unless, when)
+import Crypto.Hash (Context, SHA256, hashInit, hashUpdate, hashFinalize)
+import qualified Data.ByteString as BS
+import Data.IORef
+import System.Directory (removeFile)
+import System.IO
+
+-- A whole-stream SHA-256 snapshot, including an EOF probe at the byte budget.
+-- Register before consuming input. The exposed file has no write/resize method;
+-- only the private handle can prepare it, before its token is returned.
+snapshot :: R.Runtime -> R.Scope -> R.Token -> FilePath -> Integer -> String
+ -> IO (Either R.Problem (R.Token, Integer))
+snapshot = snapshotWith hClose
+
+snapshotWith :: (Handle -> IO ()) -> R.Runtime -> R.Scope -> R.Token -> FilePath
+ -> Integer -> String -> IO (Either R.Problem (R.Token, Integer))
+snapshotWith release runtime scope reader directory limit expected
+ | limit < 0 || limit > R.positionLimit = pure (Left (fault "invalid-bound" "invalid byte limit" "none"))
+ | length expected /= 64 || any (`notElem` ("0123456789abcdef" :: String)) expected =
+ pure (Left (fault "invalid-argument" "expected lowercase SHA-256" "none"))
+ | null directory = pure (Left (fault "invalid-argument" "empty snapshot directory" "none"))
+ | otherwise = R.withReaderLease runtime scope reader (\receive -> mask $ \restore -> do
+ slot <- newIORef Nothing
+ acquired <- R.acquire runtime scope $ do
+ (path,h) <- openBinaryTempFile directory ".tricu-bytes-"
+ (do
+ removeFile path
+ hSetBuffering h NoBuffering
+ writeIORef slot (Just h)
+ pure (R.File (R.FileOps (Just (BS.hGetSome h)) Nothing
+ (Just (hSeek h AbsoluteSeek)) (Just (hTell h)) (hFileSize h)
+ Nothing Nothing (release h)))) `onException` hClose h
+ token <- either R.resourceFault pure acquired
+ Just h <- readIORef slot
+ -- A lease protects the registered spool throughout preparation. Exceptions
+ -- leave it to scope settlement, retaining cancellation/cleanup precedence.
+ result <- restore $ R.withReaderLease runtime scope token $ \_ -> do
+ let copy !count !context = do
+ allowInterrupt
+ bytes <- receive (fromInteger (min (toInteger R.chunkLimit) (limit-count+1)))
+ if BS.null bytes then pure (count,context) else do
+ let nextCount = count + toInteger (BS.length bytes)
+ when (nextCount > limit) (R.resourceFault (fault "invalid-bound" "stream exceeds byte limit" "unknown"))
+ BS.hPut h bytes
+ next <- evaluate (hashUpdate context bytes)
+ copy nextCount next
+ (count,context) <- copy 0 (hashInit :: Context SHA256)
+ unless (show (hashFinalize context) == expected)
+ (R.resourceFault (fault "integrity-failed" "SHA-256 mismatch" "unknown"))
+ hSeek h AbsoluteSeek 0
+ pure count
+ case result of
+ Right count -> pure (token,count)
+ Left primary -> do
+ closed <- R.closeResource runtime scope token
+ R.resourceFault $ case closed of
+ Right () -> primary
+ Left cleanup -> primary { R.detail = take 240 (R.detail primary) ++ "; cleanup: " ++ take 240 (show cleanup) })
+ where fault category detail progress = R.Problem category "snapshot" detail progress
+
+-- Exact byte count, starting at the current cursor; no EOF probe. Both leases
+-- cover the transfer. A failure may follow partial writes: callers close rather
+-- than retrying or emitting a second response. No encoded byte trees are built.
+transfer :: R.Runtime -> R.Scope -> R.Token -> R.Token -> Integer -> IO (Either R.Problem ())
+transfer runtime scope reader writer count
+ | count < 0 || count > R.positionLimit = pure (Left (R.Problem "invalid-bound" "transfer" "invalid byte count" "none"))
+ | otherwise = R.withReaderLease runtime scope reader (\receive ->
+ R.withWriterLease runtime scope writer (\send -> do
+ let sendAll bytes = unless (BS.null bytes) $ do
+ n <- send bytes
+ sendAll (BS.drop n bytes)
+ copy !remaining = when (remaining > 0) $ do
+ allowInterrupt
+ bytes <- receive (fromInteger (min remaining (toInteger R.chunkLimit)))
+ when (BS.null bytes) (R.resourceFault (R.Problem "invalid-data" "transfer" "unexpected EOF" "unknown"))
+ sendAll bytes
+ copy (remaining - toInteger (BS.length bytes))
+ copy count) >>= either R.resourceFault pure)
diff --git a/src/IOV2/Driver.hs b/src/IOV2/Driver.hs
@@ -12,6 +12,7 @@ import qualified IOV2.Digest as Digest
import qualified IOV2.Filesystem as FS
import qualified IOV2.Network as TCP
import qualified IOV2.Codec as Codec
+import qualified IOV2.ByteStream as ByteStream
import qualified IOV2.UTF8 as UTF8
import qualified IOV2.Resolver as Resolver
import qualified IOV2.Clock as Clock
@@ -274,6 +275,21 @@ loop runtime marker group action frames = S.checkpoint group >> step
Just spec -> Process.launch runtime group spec >>= finish . reply
(\(task,pipes) -> ofList ([taskTree "io.v2.task" task, taskTree "io.v2.join" task] ++ map tokenTree pipes))
Nothing -> finish (bad "spawn")
+ 108 -> case payload of
+ Fork reader (Fork directory (Fork limit (Fork digest Leaf))) -> case
+ (decodeToken reader, canonicalText directory, canonicalNat 63 limit, boundedDigest digest) of
+ (Just token, Just path, Just maximum, Just expected) ->
+ ByteStream.snapshot runtime scope token path maximum expected >>= finish . reply
+ (\(snapshot,size) -> ofList [tokenTree snapshot,ofNumber size])
+ _ -> finish (bad "snapshot")
+ _ -> finish (bad "snapshot")
+ 109 -> case payload of
+ Fork reader (Fork writer (Fork count Leaf)) -> case
+ (decodeToken reader, decodeToken writer, canonicalNat 63 count) of
+ (Just input, Just output, Just size) ->
+ ByteStream.transfer runtime scope input output size >>= finish . reply (const Leaf)
+ _ -> finish (bad "transfer")
+ _ -> finish (bad "transfer")
100 -> case payload of
Leaf -> finish (ok P.capabilities)
_ -> finish (bad "capabilities")
diff --git a/src/IOV2/Protocol.hs b/src/IOV2/Protocol.hs
@@ -29,7 +29,7 @@ catalogue =
, ("resolver", [("resolve",86)])
, ("indexed-tree", [("read",87), ("hash",88), ("write",89), ("verify",95)])
, ("utf8", [("encode",96), ("decode",97)])
- , ("stream", [("read",22), ("write",23)])
+ , ("stream", [("read",22), ("write",23), ("snapshot",108), ("transfer",109)])
, ("resource", [("close",24)])
, ("stage", [("open",30), ("seal",31), ("publish",32)])
, ("digest", [("open",40), ("update",41), ("finish",42)])
diff --git a/test/IOV2ByteStreamTests.hs b/test/IOV2ByteStreamTests.hs
@@ -0,0 +1,150 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+module IOV2ByteStreamTests (ioV2ByteStreamTests) where
+
+import qualified IOV2.ByteStream as B
+import qualified IOV2.Resources as R
+import Control.Concurrent
+import Control.Exception
+import Control.Monad (forM_)
+import Crypto.Hash (hash, Digest, SHA256)
+import qualified Data.ByteString as BS
+import Data.IORef
+import System.Directory (listDirectory)
+import System.FilePath ((</>))
+import System.IO (hClose)
+import System.IO.Temp (withSystemTempDirectory)
+import Test.Tasty
+import Test.Tasty.HUnit
+
+sha :: BS.ByteString -> String
+sha bytes = show (hash bytes :: Digest SHA256)
+category wanted value = case value of
+ Left p -> R.category p @?= wanted
+ Right _ -> assertFailure ("expected " ++ wanted)
+reader runtime bytes = do
+ remaining <- newIORef bytes
+ Right token <- R.acquire runtime R.rootScope (pure (R.Reader (\n ->
+ atomicModifyIORef' remaining (\rest -> (BS.drop n rest, BS.take n rest))) (pure ())))
+ pure token
+
+ioV2ByteStreamTests = localOption (mkTimeout 30000000) $ testGroup "IO v2 native byte streams"
+ [ testCase "authenticated binary snapshots and partial writes across chunk boundaries" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> forM_ [0,1,4095,4096,4097,917828,1048576] $ \size -> do
+ let bytes = BS.pack (take size (cycle [0..255]))
+ runtime <- R.newRuntime
+ input <- reader runtime bytes
+ Right (snapshot,count) <- B.snapshot runtime R.rootScope input directory (toInteger size) (sha bytes)
+ count @?= toInteger size
+ listDirectory directory >>= (@?= [])
+ chunks <- newIORef []
+ Right output <- R.acquire runtime R.rootScope (pure (R.Writer (\chunk -> do
+ let part = BS.take 997 chunk
+ modifyIORef' chunks (part:)
+ pure (BS.length part)) (pure ())))
+ B.transfer runtime R.rootScope snapshot output count >>= (@?= Right ())
+ BS.concat . reverse <$> readIORef chunks >>= (@?= bytes)
+ R.readSome runtime R.rootScope snapshot 1 >>= (@?= Right BS.empty)
+ R.seekFile runtime R.rootScope snapshot 0 >>= (@?= Right ())
+ R.fileSize runtime R.rootScope snapshot >>= (@?= Right count)
+ R.writeSome runtime R.rootScope snapshot "x" >>= category "wrong-kind"
+ R.resizeFile runtime R.rootScope snapshot 0 >>= category "unsupported"
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ , testCase "digest and limit failures remove the spool; EOF at the limit is required" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> do
+ runtime <- R.newRuntime
+ forM_ [("abc",3,sha "xyz","integrity-failed"),("abcd",3,sha "abc","invalid-bound"),("",0,sha "x","integrity-failed")] $ \(bytes,limit,digest,wanted) -> do
+ input <- reader runtime bytes
+ counts <- R.liveCounts runtime
+ B.snapshot runtime R.rootScope input directory limit digest >>= category wanted
+ R.liveCounts runtime >>= (@?= counts)
+ listDirectory directory >>= (@?= [])
+ input <- reader runtime "abc"
+ B.snapshot runtime R.rootScope input directory (-1) (sha "abc") >>= category "invalid-bound"
+ B.snapshot runtime R.rootScope input directory 3 (replicate 64 'Z') >>= category "invalid-argument"
+ R.readSome runtime R.rootScope input 3 >>= (@?= Right "abc")
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ , testCase "source mutation after snapshot cannot change transferred bytes" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> do
+ runtime <- R.newRuntime
+ -- Injectable source bytes model in-place changes, independent of host
+ -- Handle locking of a simultaneously opened file.
+ source <- newIORef "original"
+ cursor <- newIORef 0
+ Right input <- R.acquire runtime R.rootScope (pure (R.Reader (\n -> do
+ pos <- readIORef cursor
+ bytes <- BS.take n . BS.drop pos <$> readIORef source
+ modifyIORef' cursor (+ BS.length bytes)
+ pure bytes) (pure ())))
+ Right (snapshot,count) <- B.snapshot runtime R.rootScope input directory 100 (sha "original")
+ writeIORef source "modified"
+ Right output <- R.openWriter runtime R.rootScope (directory </> "output")
+ B.transfer runtime R.rootScope snapshot output count >>= (@?= Right ())
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ BS.readFile (directory </> "output") >>= (@?= "original")
+ , testCase "mutation during copying fails authentication" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> do
+ runtime <- R.newRuntime
+ let original = BS.replicate 8192 65
+ calls <- newIORef (0 :: Int)
+ Right input <- R.acquire runtime R.rootScope (pure (R.Reader (\n -> do
+ call <- atomicModifyIORef' calls (\x -> (x+1,x))
+ pure (if call < 2 then BS.replicate n (if call == 0 then 65 else 66) else BS.empty)) (pure ())))
+ B.snapshot runtime R.rootScope input directory 8192 (sha original) >>= category "integrity-failed"
+ listDirectory directory >>= (@?= [])
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ , testCase "snapshot cancellation keeps scoped cleanup and releases source lease" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> do
+ runtime <- R.newRuntime
+ entered <- newEmptyMVar; block <- newEmptyMVar; done <- newEmptyMVar
+ Right input <- R.acquire runtime R.rootScope (pure (R.Reader (\_ -> putMVar entered () >> takeMVar block) (pure ())))
+ released <- newIORef (0 :: Int)
+ worker <- forkIO $ (B.snapshotWith (\h -> modifyIORef' released (+1) >> hClose h)
+ runtime R.rootScope input directory 4096 (sha "") >> pure ())
+ `catch` (\(_ :: AsyncException) -> pure ()) `finally` putMVar done ()
+ takeMVar entered
+ listDirectory directory >>= (@?= [])
+ killThread worker; takeMVar done
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ readIORef released >>= (@?= 1)
+ , testCase "cleanup failure preserves primary integrity diagnostic" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> do
+ runtime <- R.newRuntime
+ input <- reader runtime "bad"
+ result <- B.snapshotWith (\h -> hClose h >> ioError (userError "injected close"))
+ runtime R.rootScope input directory 3 (sha "yes")
+ category "integrity-failed" result
+ case result of Left p -> assertBool "cleanup recorded" (length (R.detail p) > length ("SHA-256 mismatch" :: String)); _ -> pure ()
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ , testCase "transfer failure and cancellation release leases without retrying" $
+ withSystemTempDirectory "tricu-bytes" $ \directory -> do
+ runtime <- R.newRuntime
+ input <- reader runtime "abcdef"
+ writes <- newIORef (0 :: Int)
+ Right broken <- R.acquire runtime R.rootScope (pure (R.Writer (\_ -> do
+ modifyIORef' writes (+1)
+ ioError (userError "peer disconnected")) (pure ())))
+ B.transfer runtime R.rootScope input broken 6 >>= category "io-error"
+ readIORef writes >>= (@?= 1)
+ R.closeResource runtime R.rootScope broken >>= (@?= Right ())
+ second <- reader runtime "abcdef"
+ entered <- newEmptyMVar; block <- newEmptyMVar; done <- newEmptyMVar
+ Right blocked <- R.acquire runtime R.rootScope (pure (R.Writer (\_ -> putMVar entered () >> takeMVar block) (pure ())))
+ worker <- forkIO $ (B.transfer runtime R.rootScope second blocked 6 >> pure ())
+ `catch` (\(_ :: AsyncException) -> pure ()) `finally` putMVar done ()
+ takeMVar entered
+ R.readSome runtime R.rootScope second 1 >>= category "busy"
+ R.writeSome runtime R.rootScope blocked "x" >>= category "busy"
+ killThread worker; takeMVar done
+ R.closeResource runtime R.rootScope second >>= (@?= Right ())
+ R.closeResource runtime R.rootScope blocked >>= (@?= Right ())
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ , testCase "transfer rejects short sources and wrong-kind targets" $ do
+ runtime <- R.newRuntime
+ input <- reader runtime "x"
+ B.transfer runtime R.rootScope input input 1 >>= category "wrong-kind"
+ chunks <- newIORef []
+ Right output <- R.acquire runtime R.rootScope (pure (R.Writer (\bytes -> modifyIORef' chunks (bytes:) >> pure (BS.length bytes)) (pure ())))
+ B.transfer runtime R.rootScope input output 2 >>= category "invalid-data"
+ readIORef chunks >>= (@?= ["x"])
+ R.closeScope runtime R.rootScope >>= (@?= [])
+ ]
diff --git a/test/IOV2ProtocolTests.hs b/test/IOV2ProtocolTests.hs
@@ -23,7 +23,7 @@ ioV2ProtocolTests = testGroup "IO v2 protocol"
let strings = ofList . map ofString
row key value = ofList [ofString key,value]
groups = [("runtime",["capabilities","inspect"]),("standard",["input","output","error"]),("file",["open-reader","open-writer","open-appender","open-update","seek","position","size","resize","sync"]),
- ("path",["stat","lstat","remove-file","rename"]),("directory",["create","remove","open","next","sync"]),("tcp",["listen","connect","accept","local-address","peer-address","shutdown-write"]),("resolver",["resolve"]),("indexed-tree",["read","hash","write","verify"]),("utf8",["encode","decode"]),("stream",["read","write"]),("resource",["close"]),("stage",["open","seal","publish"]),
+ ("path",["stat","lstat","remove-file","rename"]),("directory",["create","remove","open","next","sync"]),("tcp",["listen","connect","accept","local-address","peer-address","shutdown-write"]),("resolver",["resolve"]),("indexed-tree",["read","hash","write","verify"]),("utf8",["encode","decode"]),("stream",["read","write","snapshot","transfer"]),("resource",["close"]),("stage",["open","seal","publish"]),
("digest",["open","update","finish"]),("clock",["monotonic","sleep"]),("scheduler",["yield"]),("process",["spawn","working-directory","environment-variable"])]
expected = ofList [row "schema" (ofNumber 1), row "io-version" (ofNumber 2),
row "backend" (ofString "haskell-native"), row "authority" (ofString "os-process"),
@@ -45,7 +45,7 @@ ioV2ProtocolTests = testGroup "IO v2 protocol"
mapM_ (\(g,op) -> V.runDetailed (request g 1 op (if g == "utf8" && op == "encode" then Stem Leaf else Leaf)) >>= \outcome -> case outcome of
V.Completed (Fork Leaf (Fork (Fork _ (Fork operation _)) Leaf)) -> operation @?= ofString (g++"/"++op)
_ -> assertFailure (g++"/"++op))
- [("file","open-reader"),("file","open-writer"),("file","open-appender"),("file","open-update"),("file","seek"),("file","position"),("file","size"),("file","resize"),("file","sync"),("stream","read"),("stream","write"),
+ [("file","open-reader"),("file","open-writer"),("file","open-appender"),("file","open-update"),("file","seek"),("file","position"),("file","size"),("file","resize"),("file","sync"),("stream","read"),("stream","write"),("stream","snapshot"),("stream","transfer"),
("path","stat"),("path","lstat"),("path","remove-file"),("path","rename"),("directory","create"),("directory","remove"),("directory","open"),("directory","next"),("directory","sync"),("tcp","listen"),("tcp","connect"),("tcp","accept"),("tcp","local-address"),("tcp","peer-address"),("tcp","shutdown-write"),("resolver","resolve"),("indexed-tree","read"),("indexed-tree","hash"),("indexed-tree","write"),("indexed-tree","verify"),("utf8","encode"),("utf8","decode"),("resource","close"),("stage","open"),("stage","seal"),("stage","publish"),
("digest","open"),("digest","update"),("digest","finish"),("runtime","inspect")]
, testCase "inspection does not demand core payloads or callbacks" $
diff --git a/test/Spec.hs b/test/Spec.hs
@@ -6,6 +6,7 @@ import PortableCompilerTests (portableCompilerTests)
import ExecutableLoadingTests (executableLoadingTests)
import IODriverTests (ioDriverTests)
import IOV2UTF8Tests (ioV2UTF8Tests)
+import IOV2ByteStreamTests (ioV2ByteStreamTests)
import IOV2SyncTests (ioV2SyncTests)
import IndexedWriterTests (indexedWriterTests)
import SnapshotTests (snapshotTests)
@@ -124,6 +125,7 @@ tests = testGroup "Tricu Tests"
, ioV2ProtocolTests
, ioV2TaskTests
, ioV2UTF8Tests
+ , ioV2ByteStreamTests
, ioV2SyncTests
, indexedWriterTests
, snapshotTests
diff --git a/test/body_retention.py b/test/body_retention.py
@@ -157,7 +157,7 @@ def main():
save()
assert not errors, errors
if args.bounded_streaming:
- selected = [row for row in report['rows'] if row['phase'] == 'streaming-publish']
+ selected = [row for row in report['rows'] if row['phase'] in ['streaming-publish', 'streaming-download']]
assert selected, '--bounded-streaming needs a streaming phase'
for row in selected:
assert row['completed'] and not row['sampling_errors'], row
diff --git a/test/io-v2/README.md b/test/io-v2/README.md
@@ -336,6 +336,51 @@ DNS. The real socket source suite also resolves numeric/localhost names, checks
complete output, and exercises failed-first/successful-second source fallback,
including non-execution of a malformed later candidate and scope-bound lifetime.
+## Authenticated native byte snapshots and transfer
+
+The Haskell native host implements two revision-1 `stream` operations:
+
+- `IO.snapshotBytes reader directory limit digest` consumes the reader through
+ EOF under one read lease and returns `[snapshot size]` inside a Result. `digest`
+ is exactly 64 lowercase hexadecimal SHA-256 digits over the raw bytes. `limit`
+ is a canonical natural in `[0,2^63-1]`. The EOF probe can consume one excess byte
+ on a limit failure. A digest mismatch returns `integrity-failed`.
+- `IO.transferBytes reader writer count` transfers exactly `count` bytes from the
+ current cursor with both directional leases held. Positive short writes are
+ completed without replaying accepted prefixes. Early EOF is `invalid-data`;
+ an error can follow partial output, so callers close rather than retrying an
+ uncertain transfer. A zero count still validates and leases both handles.
+
+Snapshot preparation registers an anonymous temporary file before consuming
+input. Only authenticated snapshots are returned; they are read-only file
+resources owned by the current scope, with read/seek/position/size operations.
+They may be closed or handed off under the ordinary ownership rules. Their
+private writable handles remain native. Transfer reads the verified snapshot,
+so changes to the original file after authentication cannot change its output.
+Concurrent source mutation is accepted only if the captured bytes match the
+expected digest. This is integrity checking, not OS-process confinement.
+
+Both operations use native chunks of at most 4096 bytes and constant-size hash
+state, avoiding tree-encoded bulk byte lists. Temporary payload disk is at most
+`limit` per snapshot, plus filesystem overhead. Source readers can still block;
+operation cancellation is interruptible, with no hard deadline guarantee. Errors
+and cancellation retain scoped cleanup and diagnostic precedence. These APIs
+are native-host facilities; other backends report their own capabilities.
+
+The registry's bundle GET/HEAD path verifies before headers, logs the selected
+status, then sends the known length and verified body (HEAD sends headers only).
+Missing/corrupt/oversized objects return 404; malformed digests return 400. Logger
+failure before headers returns 500. Transfer failure after headers closes the
+connection. Whole-value artifact APIs and package metadata keep their existing
+resident-value contracts.
+
+`IOV2ByteStreamTests` covers boundaries through 1 MiB, binary/partial transfers,
+read-only snapshots, corruption, size limits, source mutation, cancellation,
+cleanup faults, early EOF and disconnected writers. `test/io_v2_http.py` exercises
+framing, configured limits and failed logging. The production-runner acceptance
+gate additionally runs `test/registry_download.py` on the actual compiler bundle
+and bounds both upload/download retention in `test/body_retention.py`.
+
## Authenticated indexed-tree reads
`IO.readIndexedTree reader directory limit size digest` consumes exactly `size`
@@ -629,6 +674,8 @@ Implemented group revisions are all **1**:
| indexed-tree | verify | `[reader limit tree]` | `[size digest]` |
| stream | read | reader, positive limit | `(data, nonempty bytes)` or `(eof, t)` |
| stream | write | writer, nonempty bytes | positive accepted count |
+| stream | snapshot | `[reader directory limit digest]` | `[snapshot size]` |
+| stream | transfer | `[reader writer count]` | unit |
| resource | close | resource | unit |
| stage | open | scope, directory | staged writer |
| stage | seal | stage | unit after flush, sync and close |
diff --git a/test/io_v2_http.py b/test/io_v2_http.py
@@ -149,7 +149,7 @@ def main():
peer.sendall(header(1048576)+b'partial')
peer.setsockopt(socket.SOL_SOCKET,socket.SO_LINGER,struct.pack('ii',1,0))
check('healthy-after-reset',request(cases[0][1]),expected(200,b'OK\n'))
- with server('downloads',6) as (root,connect,request):
+ with server('downloads',7) as (root,connect,request):
data=b'authenticated binary\0\xff';digest,path=object_path(root,data)
path.parent.mkdir(exist_ok=True);path.write_bytes(data)
wire=f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()
@@ -158,7 +158,7 @@ def main():
package=root/'packages'/'example';package.mkdir(parents=True)
(package/'v1').write_bytes(b'package bytes')
check('package-download',request(b'GET /_arboricx/package/example/v1 HTTP/1.1\r\n\r\n'),expected(200,b'package bytes','application/vnd.arboricx.package-release'))
- for size in [4096,65536]:
+ for size in [4096,65536,1048576]:
data=bytes(range(256))*(size//256);digest,path=object_path(root,data)
path.parent.mkdir(exist_ok=True);path.write_bytes(data)
check(f'download-{size}',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(200,data,'application/vnd.arboricx.bundle'))
@@ -171,14 +171,21 @@ def main():
for i in range(128):
assert request(cases[0][1])==expected(200,b'OK\n')
checks.append('128-connections-consumed-tickets')
- with server('limited',2,name='limited') as (root,connect,request):
+ with server('limited',4,name='limited') as (root,connect,request):
check('configured-upload-limit',request(header(5)+b'12345'),expected(413,b'Request body too large\n'))
assert not list((root/'objects').rglob('*.arboricx'))
check('configured-upload-boundary',request(header(4)+b'1234'),expected(201,hashlib.sha256(b'1234').hexdigest().encode()))
- with server('logger',1,name='badLogger') as (root,connect,request):
+ for data,status in [(b'1234',200),(b'12345',404)]:
+ digest,path=object_path(root,data)
+ path.parent.mkdir(exist_ok=True);path.write_bytes(data)
+ body=data if status==200 else b'Artifact unavailable\n'
+ mime='application/vnd.arboricx.bundle' if status==200 else 'text/plain; charset=utf-8'
+ check(f'configured-download-{len(data)}',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(status,body,mime))
+ with server('logger',2,name='badLogger') as (root,connect,request):
data=b'published before logger failure';digest,path=object_path(root,data)
check('logger-failure',request(header(len(data))+data),expected(500,b'Internal Server Error\n'))
assert path.read_bytes()==data
+ check('download-logger-failure',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(500,b'Internal Server Error\n'))
# Two pending streams occupy exactly two admission slots. A third connection
# may enter the kernel backlog but cannot create another stage until a ticket
# is consumed. FIFO deliberately waits for the oldest, not any completed child.
diff --git a/test/registry_download.py b/test/registry_download.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Production runner: compiler download latency, native memory and error controls.
+
+Localhost observations include request parsing, authentication and socket output;
+these are regression ceilings, not predictions of production network latency.
+"""
+import argparse
+import concurrent.futures
+import hashlib
+import json
+from pathlib import Path
+import signal
+import socket
+import struct
+import subprocess
+import tempfile
+import threading
+import time
+from process_rss import ChildRSS
+from rts_metrics import parse_rts_stats
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('runner', type=Path)
+ parser.add_argument('compiler', type=Path)
+ parser.add_argument('--report', type=Path, required=True)
+ args = parser.parse_args()
+ data = args.compiler.read_bytes()
+ digest = hashlib.sha256(data).hexdigest()
+ route = '/_arboricx/bundle/hash/' + digest
+ rows = []
+ with tempfile.TemporaryDirectory(prefix='registry-download-') as temporary:
+ root = Path(temporary)
+ obj = root / 'store' / 'objects' / digest[:3] / (digest + '.arboricx')
+ obj.parent.mkdir(parents=True)
+ obj.write_bytes(data)
+ stopped = threading.Event()
+ peak = [0]
+ errors = []
+ with (root / 'server.log').open('w+') as log:
+ import os
+ proc = subprocess.Popen([str(args.runner.resolve())], cwd=root, stdout=log, stderr=log,
+ env=dict(os.environ, GHCRTS='-T -s'))
+ def monitor():
+ try:
+ meter = ChildRSS(proc)
+ while not stopped.is_set():
+ value = meter.sample()
+ if value is not None:
+ peak[0] = max(peak[0], value)
+ if value > 128 * 1024:
+ raise AssertionError('128 MiB RSS guard exceeded')
+ stopped.wait(.01)
+ except Exception as error:
+ errors.append(repr(error))
+ proc.kill()
+ watcher = threading.Thread(target=monitor, daemon=True)
+ watcher.start()
+ def request(path=route, method='GET', status=200, body=data):
+ begin = time.monotonic()
+ with socket.create_connection(('127.0.0.1', 8080), timeout=10) as peer:
+ peer.sendall(f'{method} {path} HTTP/1.1\r\nHost: local\r\n\r\n'.encode())
+ first = peer.recv(65536)
+ first_byte = time.monotonic() - begin
+ chunks = [first]
+ while chunk := peer.recv(65536):
+ chunks.append(chunk)
+ total = time.monotonic() - begin
+ headers, received = b''.join(chunks).split(b'\r\n\r\n', 1)
+ assert int(headers.split()[1]) == status, headers
+ assert received == (b'' if method == 'HEAD' else body)
+ assert f'Content-Length: {len(body)}'.encode() in headers
+ return dict(first_byte_seconds=first_byte, total_seconds=total)
+ try:
+ for _ in range(100):
+ assert proc.poll() is None, 'server exited'
+ try:
+ request('/_arboricx/health', body=b'OK\n')
+ break
+ except OSError:
+ time.sleep(.1)
+ else:
+ raise AssertionError('startup timeout')
+ for _ in range(3):
+ rows.append(request())
+ request(method='HEAD')
+ with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+ rows.extend(pool.map(lambda _: request(), range(8)))
+ # Corruption is rejected before 200 headers; HEAD authenticates too.
+ obj.write_bytes(b'X' + data[1:])
+ request(status=404, body=b'Artifact unavailable\n')
+ request(method='HEAD', status=404, body=b'Artifact unavailable\n')
+ obj.write_bytes(data + b'x')
+ request(status=404, body=b'Artifact unavailable\n')
+ obj.write_bytes(data)
+ request('/_arboricx/bundle/hash/' + '0' * 64, status=404, body=b'Artifact unavailable\n')
+ # Reset after headers while the body may still be in flight.
+ with socket.create_connection(('127.0.0.1', 8080), timeout=10) as peer:
+ peer.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
+ peer.sendall(f'GET {route} HTTP/1.1\r\n\r\n'.encode())
+ assert peer.recv(1) == b'H'
+ peer.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+ request('/_arboricx/health', body=b'OK\n')
+ rows.append(request())
+ assert list((root / 'store' / 'tmp').iterdir()) == []
+ assert all(row['total_seconds'] < 5 for row in rows), rows
+ finally:
+ if proc.poll() is None:
+ proc.send_signal(signal.SIGINT)
+ proc.wait(timeout=15)
+ stopped.set()
+ watcher.join()
+ log.seek(0)
+ stats = parse_rts_stats(log.read())
+ assert not errors, errors
+ assert stats['maximum_live_bytes'] < 16 * 1024 * 1024, stats
+ report = dict(sha256=digest, bytes=len(data), requests=rows, peak_rss_kib=peak[0], rts=stats)
+ args.report.write_text(json.dumps(report, indent=2) + '\n')
+ print(json.dumps(report, indent=2))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/tricu.cabal b/tricu.cabal
@@ -1,7 +1,7 @@
cabal-version: 1.12
name: tricu
-version: 2026.10.19
+version: 2026.10.24
description: A language for exploring Tree Calculus
author: James Eversole
maintainer: james@eversole.co
@@ -96,6 +96,7 @@ executable tricu
IOV2.Staging
IOV2.Digest
IOV2.Codec
+ IOV2.ByteStream
IOV2.Clock
IOV2.Process
IOV2.Context
@@ -225,6 +226,7 @@ test-suite tricu-tests
IOV2.Staging
IOV2.Digest
IOV2.Codec
+ IOV2.ByteStream
IOV2.Clock
IOV2.Process
IOV2.Context
@@ -258,6 +260,7 @@ test-suite tricu-tests
SnapshotTests
IndexedWriterTests
IOV2SyncTests
+ IOV2ByteStreamTests
IOV2UTF8Tests
IOV2NetworkTests
IOV2FilesystemTests
