Stream object transfers with shared byte throttling

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test check-dependencies check-fixture-lib check-kind-matrix check-storage-foundation check-auth check-app app-up app-down check-frontend build-frontend
+.PHONY: test check-dependencies check-fixture-lib check-kind-matrix check-storage-foundation check-auth check-app check-streaming app-up app-down check-frontend build-frontend
 
 test:
 	cabal test --offline nuthatch-unit
@@ -21,6 +21,9 @@ check-auth:
 check-app:
 	./tests/kind/wp05-run.sh
 
+check-streaming:
+	./tests/kind/wp07-run.sh
+
 app-up:
 	NUTHATCH_APP_INTERACTIVE=1 ./tests/kind/wp05-run.sh
 

nuthatch.cabal

@@ -107,6 +107,7 @@ library
     Nuthatch.Redis
     Nuthatch.S3
     Nuthatch.Session
+    Nuthatch.Streaming
     Nuthatch.Types
     Nuthatch.Ui
 

src/Nuthatch/Api.hs

@@ -431,6 +431,7 @@ type DownloadSuccess =
   Headers
     '[ Header "Content-Disposition" Text
      , Header "X-Content-Type-Options" Text
+     , Header "Content-Length" Integer
      ]
     (SourceIO BS.ByteString)
 

src/Nuthatch/App.hs

@@ -9,15 +9,11 @@ module Nuthatch.App
   ) where
 
 import           Control.Concurrent (threadDelay)
-import           Control.Exception (SomeException, try)
-import           Control.Monad (forever, unless, void, when)
+import           Control.Exception (SomeException, throwIO, try)
+import           Control.Monad (forever, void, when)
 import           Control.Monad.IO.Class (liftIO)
-import           Control.Monad.Trans.Except (runExceptT)
-import           Crypto.Hash (Digest, SHA256, hashlazy)
 import qualified Data.Aeson as A
-import qualified Data.ByteArray as BA
 import qualified Data.ByteString as BS
-import qualified Data.ByteString.Base16 as B16
 import qualified Data.ByteString.Lazy as BL
 import           Data.Either (isRight)
 import           Data.Maybe (fromMaybe, listToMaybe)
@@ -43,19 +39,33 @@ import qualified Nuthatch.Db as Db
 import           Nuthatch.Oidc
 import           Nuthatch.Paths
 import           Nuthatch.Protocol
-import           Nuthatch.Redis (Redis, cacheGet, cacheInvalidatePrefixes, cacheSet, throttleTake)
+import           Nuthatch.Redis
+  ( Redis
+  , cacheGet
+  , cacheInvalidatePrefixes
+  , cacheSet
+  , throttleCheck
+  , throttleTake
+  )
 import           Nuthatch.S3
   ( ListPage (..)
   , ObjectInfo (..)
+  , PutObjectError (..)
   , S3
+  , S3Error (..)
   , deleteObject
-  , getObjectBytes
+  , getObjectSource
   , headObject
   , listObjects
-  , putObject
+  , observeObjectFully
+  , putObjectStream
   , renderS3Error
   )
 import           Nuthatch.Session
+import           Nuthatch.Streaming
+  ( BandwidthError (..)
+  , UploadStreamError (..)
+  )
 import           Nuthatch.Types
 import           Nuthatch.Ui (browsePage, browserScript)
 import           Servant
@@ -63,7 +73,7 @@ import           Servant.Server.Experimental.Auth
   ( AuthHandler
   , mkAuthHandler
   )
-import           Servant.Types.SourceT (runSourceT, source)
+import           Servant.Types.SourceT (SourceT)
 import           Web.Cookie
   ( defaultSetCookie
   , parseCookies
@@ -306,17 +316,9 @@ handleDownloadRequest env session root path = do
       (DownloadObjectStoreFailed (renderS3Error message))
     Right Nothing -> throwOperationError err404 DownloadNotFound
     Right (Just value) -> pure value
-  throttle <- liftIO $
-    throttleTake (envRedis env) (envConfig env) (suUserId user) (oiSize info)
-  case throttle of
-    Left _ -> throwOperationError err503 DownloadThrottleUnavailable
-    Right False -> throwOperationError err429 DownloadBandwidthExceeded
-    Right True -> pure ()
-  result <- liftIO (getObjectBytes (envS3 env) key)
-  body <- either
-    (throwOperationError status502Error . DownloadObjectStoreFailed . renderS3Error)
-    pure
-    result
+  requireDownloadBandwidth env user (oiSize info)
+  let inspect = inspectBandwidth env user
+      body = getObjectSource (envS3 env) key (transferPieceSize env) inspect
   liftIO $ audit "download" "ok" (Just (unUserId (suUserId user)))
     [ ("root", A.toJSON (rootLabel root))
     , ("bytes", A.toJSON (oiSize info))
@@ -324,7 +326,8 @@ handleDownloadRequest env session root path = do
   let filename = sanitizeFilename (basename path)
   pure $
     addHeader ("attachment; filename=\"" <> filename <> "\"") $
-      addHeader "nosniff" (source (BL.toChunks body))
+      addHeader "nosniff" $
+        addHeader (oiSize info) body
 
 handleUploadRequest
   :: Env
@@ -339,31 +342,16 @@ handleUploadRequest
   -> Handler UploadResponse
 handleUploadRequest env session root path operation declaredSize contentIdentity providedCsrf bodySource = do
   requireCsrf UploadCsrfRejected session providedCsrf
-  chunksResult <- liftIO (runExceptT (runSourceT bodySource))
-  chunks <- either
-    (throwOperationError err400 . UploadInvalidBodyStream . T.pack)
-    pure
-    chunksResult
   let user = sessionUser session
-      body = BL.fromChunks chunks
       size = unDeclaredSize declaredSize
-      actualSize = toInteger (BL.length body)
       configuredMaximum = cfgMaxObjectBytes (envConfig env)
-  unless (actualSize == size) $
-    throwOperationError err400 UploadSizeMismatch
   when (size > configuredMaximum) $
     throwOperationError err413 UploadTooLarge
-  unless (contentSha256 body == contentIdentity) $
-    throwOperationError err400 UploadContentIdentityMismatch
-  throttle <- liftIO $
-    throttleTake (envRedis env) (envConfig env) (suUserId user) size
-  case throttle of
-    Left _ -> throwOperationError err503 UploadThrottleUnavailable
-    Right False -> throwOperationError err429 UploadBandwidthExceeded
-    Right True -> pure ()
-  outcome <- liftIO (runPut env user root path operation size contentIdentity body)
+  requireUploadBandwidth env user size
+  outcome <- liftIO $
+    runPut env user root path operation size contentIdentity bodySource
   result <- either
-    (throwOperationError err409 . UploadMutationRejected)
+    throwPutFailure
     pure
     outcome
   liftIO $ audit "put" (mutationAuditResult result) (Just (unUserId (suUserId user)))
@@ -376,6 +364,48 @@ handleUploadRequest env session root path operation declaredSize contentIdentity
     , uploadResponseResult = result
     }
 
+requireDownloadBandwidth :: Env -> SessionUser -> Integer -> Handler ()
+requireDownloadBandwidth env user size = do
+  available <- liftIO $
+    throttleCheck
+      (envRedis env)
+      (envConfig env)
+      (suUserId user)
+      (min size (toInteger (transferPieceSize env)))
+  case available of
+    Left _ -> throwOperationError err503 DownloadThrottleUnavailable
+    Right False -> throwOperationError err429 DownloadBandwidthExceeded
+    Right True -> pure ()
+
+requireUploadBandwidth :: Env -> SessionUser -> Integer -> Handler ()
+requireUploadBandwidth env user size = do
+  available <- liftIO $
+    throttleCheck
+      (envRedis env)
+      (envConfig env)
+      (suUserId user)
+      (min size (toInteger (transferPieceSize env)))
+  case available of
+    Left _ -> throwOperationError err503 UploadThrottleUnavailable
+    Right False -> throwOperationError err429 UploadBandwidthExceeded
+    Right True -> pure ()
+
+inspectBandwidth :: Env -> SessionUser -> BS.ByteString -> IO ()
+inspectBandwidth env user chunk = do
+  result <- throttleTake
+    (envRedis env)
+    (envConfig env)
+    (suUserId user)
+    (toInteger (BS.length chunk))
+  case result of
+    Left _ -> throwIO BandwidthUnavailable
+    Right False -> throwIO BandwidthExceeded
+    Right True -> pure ()
+
+transferPieceSize :: Env -> Int
+transferPieceSize env =
+  fromInteger (max 1 (min 65536 (cfgThrottleBurstBytes (envConfig env))))
+
 handleDeleteRequest
   :: Env
   -> SessionContext
@@ -495,8 +525,8 @@ runPut
   -> OperationId
   -> Integer
   -> ContentSha256
-  -> BL.ByteString
-  -> IO (Either Text MutationResult)
+  -> SourceT IO BS.ByteString
+  -> IO (Either PutFailure MutationResult)
 runPut env user root rel op size sha body = do
   ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
   let uid = suUserId user
@@ -509,12 +539,57 @@ runPut env user root rel op size sha body = do
       (cfgMaxObjectBytes (envConfig env))
   case outcome of
     AdmitConflict -> pure (Right MutationConflict)
-    AdmitRejected reason _ -> pure (Left reason)
+    AdmitRejected reason _ -> pure (Left (PutRejected reason))
     AdmitReplay MutationPending _ -> pure (Right MutationPending)
     AdmitReplay result _ -> pure (Right result)
-    AdmitPrepared rec -> dispatchAndFinalize env user root rec $ do
-      _ <- putObject (envS3 env) key size (Just sha) body
-      pure ()
+    AdmitPrepared rec ->
+      dispatchAndFinalize
+        env
+        user
+        root
+        rec
+        PutSettlementFailed
+        (putObjectStream
+          (envS3 env)
+          key
+          size
+          (Just sha)
+          (transferPieceSize env)
+          (inspectBandwidth env user)
+          body
+          >>= pure . either (Left . PutDispatchFailed) Right)
+
+data PutFailure
+  = PutRejected Text
+  | PutDispatchFailed PutObjectError
+  | PutSettlementFailed
+  deriving stock (Eq, Show)
+
+throwPutFailure :: PutFailure -> Handler a
+throwPutFailure failure =
+  case failure of
+    PutRejected reason ->
+      throwOperationError err409 (UploadMutationRejected reason)
+    PutDispatchFailed (PutObjectStreamError streamError) ->
+      case streamError of
+        UploadStreamInvalid message ->
+          throwOperationError err400 (UploadInvalidBodyStream (T.pack message))
+        UploadStreamSizeMismatch ->
+          throwOperationError err400 UploadSizeMismatch
+        UploadStreamIdentityMismatch ->
+          throwOperationError err400 UploadContentIdentityMismatch
+    PutDispatchFailed (PutObjectBandwidthError bandwidthError) ->
+      case bandwidthError of
+        BandwidthUnavailable ->
+          throwOperationError err503 UploadThrottleUnavailable
+        BandwidthExceeded ->
+          throwOperationError err429 UploadBandwidthExceeded
+    PutDispatchFailed (PutObjectStoreError err) ->
+      throwOperationError err409
+        (UploadMutationRejected (renderS3Error err))
+    PutSettlementFailed ->
+      throwOperationError err409
+        (UploadMutationRejected "mutation failed after store settlement")
 
 runDelete
   :: Env
@@ -538,9 +613,15 @@ runDelete env user root rel op = do
     AdmitRejected reason _ -> pure (Left reason)
     AdmitReplay MutationPending _ -> pure (Right MutationPending)
     AdmitReplay result _ -> pure (Right result)
-    AdmitPrepared rec -> dispatchAndFinalize env user root rec $ do
-      _ <- deleteObject (envS3 env) key
-      pure ()
+    AdmitPrepared rec ->
+      dispatchAndFinalize
+        env
+        user
+        root
+        rec
+        "mutation failed after store settlement"
+        (deleteObject (envS3 env) key
+          >>= pure . either (Left . renderS3Error) Right)
 
 deletePrefix :: Env -> SessionUser -> Root -> Text -> IO (Either Text Int)
 deletePrefix env user root prefix = do
@@ -596,9 +677,10 @@ dispatchAndFinalize
   -> SessionUser
   -> Root
   -> MutationRecord
-  -> IO ()
-  -> IO (Either Text MutationResult)
-dispatchAndFinalize env user root rec dispatch = do
+  -> failure
+  -> IO (Either failure ())
+  -> IO (Either failure MutationResult)
+dispatchAndFinalize env user root rec settlementFailure dispatch = do
   prepared <- withTransaction (envDb env) $ \conn -> do
     moved <- markApplying conn rec
     if not moved
@@ -607,7 +689,7 @@ dispatchAndFinalize env user root rec dispatch = do
   if not prepared
     then pure (Right MutationPending)
     else do
-      _ <- try dispatch :: IO (Either SomeException ())
+      dispatchResult <- dispatch
       observed <- observeObject (envS3 env) (mrObjectKey rec)
       finalized <- withTransaction (envDb env) $ \conn -> do
         settled <- markExternalSettled conn rec
@@ -620,23 +702,17 @@ dispatchAndFinalize env user root rec dispatch = do
             (listCacheAncestorPrefixes (mrRelativePath rec))
           pure (Right MutationSucceeded)
         Just MutationFailed ->
-          pure (Left "mutation failed after store settlement")
+          pure (Left (either id (const settlementFailure) dispatchResult))
         Just other -> pure (Right other)
         Nothing -> pure (Right MutationPending)
 
 observeObject :: S3 -> ObjectKey -> IO ObjectObservation
 observeObject s3 key = do
-  result <- headObject s3 key
+  result <- observeObjectFully s3 key
   case result of
-    Right (Just info@ObjectInfo { oiSha256 = Just _ }) ->
+    Right info ->
       pure (ObjectPresent (oiSize info) (oiSha256 info))
-    Right (Just _) -> do
-      body <- getObjectBytes s3 key
-      pure $ case body of
-        Right bytes ->
-          ObjectPresent (toInteger (BL.length bytes)) (Just (contentSha256 bytes))
-        Left err -> ObjectUnavailable (renderS3Error err)
-    Right Nothing -> pure ObjectAbsent
+    Left S3NotFound -> pure ObjectAbsent
     Left err -> pure (ObjectUnavailable (renderS3Error err))
 
 listEntries :: Env -> SessionUser -> Root -> Text -> IO (Either Text [ListEntry])
@@ -781,10 +857,6 @@ recoverOne env rec =
       pure ()
     _ -> pure ()
 
-contentSha256 :: BL.ByteString -> ContentSha256
-contentSha256 body =
-  ContentSha256 . TE.decodeUtf8 . B16.encode . BA.convert $
-    (hashlazy body :: Digest SHA256)
 
 lastSeg :: Text -> Text
 lastSeg t =

src/Nuthatch/Config.hs

@@ -10,6 +10,7 @@ import           Data.Maybe (fromMaybe)
 import           Data.Text (Text)
 import qualified Data.Text as T
 import qualified Data.Text.Encoding as TE
+import           Nuthatch.Streaming (maximumMultipartBytes)
 import           System.Environment (lookupEnv)
 import           Text.Read (readMaybe)
 
@@ -102,6 +103,12 @@ loadConfig = do
   maxObject <- envInteger "NUTHATCH_MAX_OBJECT_BYTES" (100 * 1024 * 1024)
   throttleRate <- envInteger "NUTHATCH_THROTTLE_BYTES_PER_SEC" (10 * 1024 * 1024)
   throttleBurst <- envInteger "NUTHATCH_THROTTLE_BURST_BYTES" (20 * 1024 * 1024)
+  when (maxObject <= 0 || maxObject > maximumMultipartBytes) $
+    fail "NUTHATCH_MAX_OBJECT_BYTES exceeds the supported streaming range"
+  when (throttleRate <= 0 || throttleRate > luaExactIntegerMax) $
+    fail "NUTHATCH_THROTTLE_BYTES_PER_SEC exceeds the exact throttle range"
+  when (throttleBurst <= 0 || throttleBurst > luaExactIntegerMax) $
+    fail "NUTHATCH_THROTTLE_BURST_BYTES exceeds the exact throttle range"
   listTtl <- envInteger "NUTHATCH_LIST_CACHE_TTL_SEC" 30
   searchMax <- fromInteger <$> envInteger "NUTHATCH_SEARCH_MAX_KEYS" 500
   recoveryGrace <- envInteger "NUTHATCH_RECOVERY_GRACE_SEC" 90
@@ -143,6 +150,9 @@ loadConfig = do
     , cfgRecoveryGraceSec = recoveryGrace
     }
 
+luaExactIntegerMax :: Integer
+luaExactIntegerMax = 9007199254740991
+
 splitHostPort :: String -> (String, String)
 splitHostPort s =
   case break (== ':') s of

src/Nuthatch/Redis.hs

@@ -4,6 +4,7 @@ module Nuthatch.Redis
   , cacheGet
   , cacheSet
   , cacheInvalidatePrefixes
+  , throttleCheck
   , throttleTake
   ) where
 
@@ -13,7 +14,6 @@ import qualified Data.ByteString.Char8 as BS8
 import           Data.Text (Text)
 import qualified Data.Text as T
 import qualified Data.Text.Encoding as TE
-import           Data.Time.Clock.POSIX (getPOSIXTime)
 import           Database.Redis
   ( ConnectInfo (..)
   , Connection
@@ -69,9 +69,21 @@ cacheInvalidatePrefixes conn uid root prefixes = do
             :: IO (Either SomeException (Either Reply Integer))
       pure ()
 
+throttleCheck :: Redis -> Config -> UserId -> Integer -> IO (Either Text Bool)
+throttleCheck conn cfg uid bytes =
+  throttleRequest conn cfg uid bytes False
+
 throttleTake :: Redis -> Config -> UserId -> Integer -> IO (Either Text Bool)
-throttleTake conn cfg uid bytes = do
-  now <- floor <$> getPOSIXTime
+throttleTake conn cfg uid bytes = throttleRequest conn cfg uid bytes True
+
+throttleRequest
+  :: Redis
+  -> Config
+  -> UserId
+  -> Integer
+  -> Bool
+  -> IO (Either Text Bool)
+throttleRequest conn cfg uid bytes consume = do
   let key = TE.encodeUtf8 ("nuthatch:throttle:" <> unUserId uid)
       rate = cfgThrottleBytesPerSec cfg
       burst = cfgThrottleBurstBytes cfg
@@ -80,8 +92,11 @@ throttleTake conn cfg uid bytes = do
           [ "local key = KEYS[1]"
           , "local rate = tonumber(ARGV[1])"
           , "local burst = tonumber(ARGV[2])"
-          , "local now = tonumber(ARGV[3])"
-          , "local need = tonumber(ARGV[4])"
+          , "local need = tonumber(ARGV[3])"
+          , "local consume = tonumber(ARGV[4])"
+          , "local clock = redis.call('TIME')"
+          , "local now = tonumber(clock[1]) + tonumber(clock[2]) / 1000000"
+          , "local ttl = math.max(1, math.ceil(burst / rate))"
           , "local data = redis.call('HMGET', key, 'tokens', 'ts')"
           , "local tokens = tonumber(data[1])"
           , "local ts = tonumber(data[2])"
@@ -91,20 +106,20 @@ throttleTake conn cfg uid bytes = do
           , "tokens = math.min(burst, tokens + delta * rate)"
           , "if tokens < need then"
           , "  redis.call('HMSET', key, 'tokens', tokens, 'ts', now)"
-          , "  redis.call('EXPIRE', key, 3600)"
+          , "  redis.call('EXPIRE', key, ttl)"
           , "  return 0"
           , "end"
-          , "tokens = tokens - need"
+          , "if consume == 1 then tokens = tokens - need end"
           , "redis.call('HMSET', key, 'tokens', tokens, 'ts', now)"
-          , "redis.call('EXPIRE', key, 3600)"
+          , "redis.call('EXPIRE', key, ttl)"
           , "return 1"
           ]
   result <- try (runRedis conn $
     R.eval script [key]
       [ BS8.pack (show rate)
       , BS8.pack (show burst)
-      , BS8.pack (show (now :: Integer))
       , BS8.pack (show bytes)
+      , if consume then "1" else "0"
       ]) :: IO (Either SomeException (Either Reply Integer))
   pure $ case result of
     Left _ -> Left "redis unavailable"

src/Nuthatch/S3.hs

@@ -1,12 +1,16 @@
 module Nuthatch.S3
   ( S3
   , S3Error (..)
+  , PutObjectError (..)
   , ObjectInfo (..)
   , ListPage (..)
   , newS3
   , putObject
+  , putObjectStream
   , getObjectBytes
+  , getObjectSource
   , headObject
+  , observeObjectFully
   , deleteObject
   , listObjects
   , renderS3Error
@@ -16,8 +20,24 @@ import qualified Aws
 import qualified Aws.Aws as AwsRun
 import qualified Aws.Core as AwsCore
 import qualified Aws.S3 as S3
+import           Control.Concurrent.Async (withAsync)
+import           Control.Concurrent.STM
+  ( atomically
+  , newTBQueueIO
+  , readTBQueue
+  , writeTBQueue
+  )
 import           Control.Applicative ((<|>))
-import           Control.Exception (SomeException, fromException, try)
+import           Control.Exception
+  ( SomeAsyncException
+  , SomeException
+  , fromException
+  , onException
+  , throwIO
+  , try
+  )
+import           Control.Monad (void)
+import           Control.Monad.IO.Class (liftIO)
 import           Control.Monad.Trans.Resource (runResourceT)
 import           Crypto.Hash
   ( Context
@@ -32,6 +52,7 @@ import qualified Data.ByteString.Char8 as BS8
 import qualified Data.ByteString.Lazy as BL
 import qualified Data.Conduit as C
 import qualified Data.Conduit.List as CL
+import           Data.IORef
 import           Data.Text (Text)
 import qualified Data.Text as T
 import qualified Data.Text.Encoding as TE
@@ -40,7 +61,18 @@ import           Network.HTTP.Client.TLS (tlsManagerSettings)
 import           Network.HTTP.Types.Status (statusCode)
 import           Network.URI (URI (..), URIAuth (..), parseURI)
 import           Nuthatch.Config (Config (..))
+import           Nuthatch.Streaming
+  ( BandwidthError (..)
+  , UploadStreamError (..)
+  , consumeUploadParts
+  , maximumMultipartBytes
+  , multipartPartBytes
+  )
 import           Nuthatch.Types
+import           Servant.Types.SourceT
+  ( SourceT (..)
+  , StepT (..)
+  )
 import           Text.Read (readMaybe)
 
 data S3 = S3
@@ -57,6 +89,12 @@ data S3Error
   | S3ResponseError
   deriving stock (Eq, Show)
 
+data PutObjectError
+  = PutObjectStoreError S3Error
+  | PutObjectStreamError UploadStreamError
+  | PutObjectBandwidthError BandwidthError
+  deriving stock (Eq, Show)
+
 data ObjectInfo = ObjectInfo
   { oiSize   :: Integer
   , oiSha256 :: Maybe ContentSha256
@@ -127,6 +165,110 @@ newS3 cfg = do
     , s3Bucket = cfgS3Bucket cfg
     }
 
+putObjectStream
+  :: S3
+  -> ObjectKey
+  -> Integer
+  -> Maybe ContentSha256
+  -> Int
+  -> (BS8.ByteString -> IO ())
+  -> SourceT IO BS8.ByteString
+  -> IO (Either PutObjectError ())
+putObjectStream store (ObjectKey key) size sha pieceSize inspect body =
+  if size > maximumMultipartBytes
+    then pure (Left (PutObjectStoreError S3ResponseError))
+    else case sha of
+      Nothing -> pure (Left (PutObjectStoreError S3ResponseError))
+      Just identity
+        | size == 0 -> do
+          streamed <- trySynchronous
+            (consumeUploadParts
+              size identity pieceSize multipartPartBytes inspect body
+              (const (throwIO UploadStreamSizeMismatch)))
+          case streamed of
+            Left exception -> pure (Left (classifyPutException exception))
+            Right () -> putEmptyObject store key sha
+        | otherwise -> do
+          initiated <- trySynchronous $ runMemory store $
+            (S3.postInitiateMultipartUpload (s3Bucket store) key)
+              { S3.imuContentType = Just "application/octet-stream"
+              , S3.imuMetadata =
+                  [("nuthatch-sha256", unContentSha256 identity)]
+              }
+          case initiated of
+            Left exception ->
+              pure (Left (PutObjectStoreError (classifyException exception)))
+            Right response -> uploadMultipart
+              store key size identity pieceSize inspect body
+              (S3.imurUploadId response)
+
+putEmptyObject
+  :: S3
+  -> Text
+  -> Maybe ContentSha256
+  -> IO (Either PutObjectError ())
+putEmptyObject store key sha = do
+  result <- trySynchronous $ runMemory store $
+    (S3.putObject (s3Bucket store) key (HTTP.RequestBodyBS BS8.empty))
+      { S3.poContentType = Just "application/octet-stream"
+      , S3.poMetadata = maybe []
+          (\identity ->
+            [("nuthatch-sha256", unContentSha256 identity)])
+          sha
+      }
+  pure $ either
+    (Left . PutObjectStoreError . classifyException)
+    (const (Right ()))
+    result
+
+uploadMultipart
+  :: S3
+  -> Text
+  -> Integer
+  -> ContentSha256
+  -> Int
+  -> (BS8.ByteString -> IO ())
+  -> SourceT IO BS8.ByteString
+  -> Text
+  -> IO (Either PutObjectError ())
+uploadMultipart store key size identity pieceSize inspect body uploadId = do
+  nextPartRef <- newIORef (1 :: Integer)
+  etagsRef <- newIORef []
+  let abort = void (try (runMemory store $
+        S3.postAbortMultipartUpload (s3Bucket store) key uploadId)
+        :: IO (Either SomeException S3.AbortMultipartUploadResponse))
+      upload = do
+        consumeUploadParts
+          size identity pieceSize multipartPartBytes inspect body $ \part -> do
+            partNumber <- readIORef nextPartRef
+            response <- runMemory store $
+              S3.uploadPart
+                (s3Bucket store)
+                key
+                partNumber
+                uploadId
+                (HTTP.RequestBodyBS part)
+            modifyIORef' etagsRef ((partNumber, S3.uprETag response) :)
+            writeIORef nextPartRef (partNumber + 1)
+        etags <- reverse <$> readIORef etagsRef
+        void $ runMemory store $
+          S3.postCompleteMultipartUpload
+            (s3Bucket store)
+            key
+            uploadId
+            etags
+  result <- trySynchronous (upload `onException` abort)
+  case result of
+    Right () -> pure (Right ())
+    Left exception -> pure (Left (classifyPutException exception))
+
+classifyPutException :: SomeException -> PutObjectError
+classifyPutException exception =
+  case (fromException exception, fromException exception) of
+    (Just streamError, _) -> PutObjectStreamError streamError
+    (_, Just bandwidthError) -> PutObjectBandwidthError bandwidthError
+    _ -> PutObjectStoreError (classifyException exception)
+
 putObject
   :: S3
   -> ObjectKey
@@ -154,6 +296,61 @@ getObjectBytes store (ObjectKey key) = do
   result <- getObjectMemory store key
   pure $ result >>= \(_, body) -> Right body
 
+data DownloadEvent
+  = DownloadChunk BS8.ByteString
+  | DownloadFailed S3Error
+  | DownloadFinished
+
+getObjectSource
+  :: S3
+  -> ObjectKey
+  -> Int
+  -> (BS8.ByteString -> IO ())
+  -> SourceT IO BS8.ByteString
+getObjectSource store (ObjectKey key) pieceSize inspect =
+  SourceT $ \consume -> do
+    queue <- newTBQueueIO 2
+    withAsync (produce queue) $ \_ ->
+      consume (consumeQueue queue)
+  where
+    produce queue = do
+      result <- trySynchronous (runResourceT $ do
+        response <- AwsRun.pureAws
+          (s3Configuration store)
+          (s3Service store)
+          (s3Manager store)
+          (S3.getObject (s3Bucket store) key)
+        C.runConduit $
+          HTTP.responseBody (S3.gorResponse response)
+            C..| C.awaitForever
+              (\chunk -> mapM_
+                (liftIO . atomically . writeTBQueue queue . DownloadChunk)
+                (splitPieces chunk)))
+        :: IO (Either SomeException ())
+      atomically $ writeTBQueue queue $ case result of
+        Left exception -> DownloadFailed (classifyException exception)
+        Right () -> DownloadFinished
+
+    consumeQueue queue = Effect $ atomically (readTBQueue queue) >>= \case
+      DownloadChunk chunk -> trySynchronous (inspect chunk) >>= \case
+        Left exception ->
+          pure (Error (T.unpack (renderDownloadException exception)))
+        Right () -> pure (Yield chunk (consumeQueue queue))
+      DownloadFailed err -> pure (Error (T.unpack (renderS3Error err)))
+      DownloadFinished -> pure Stop
+
+    splitPieces chunk
+      | BS8.null chunk = []
+      | otherwise =
+          let (piece, rest) = BS8.splitAt pieceSize chunk
+          in piece : splitPieces rest
+
+    renderDownloadException exception =
+      case fromException exception of
+        Just BandwidthUnavailable -> "bandwidth throttle unavailable"
+        Just BandwidthExceeded -> "bandwidth limit exceeded"
+        Nothing -> renderS3Error (classifyException exception)
+
 headObject :: S3 -> ObjectKey -> IO (Either S3Error (Maybe ObjectInfo))
 headObject store (ObjectKey key) = do
   result <- try (runMemory store (S3.headObject (s3Bucket store) key))
@@ -162,9 +359,37 @@ headObject store (ObjectKey key) = do
     Left err -> pure (Left (classifyException err))
     Right (S3.HeadObjectMemoryResponse Nothing) -> pure (Right Nothing)
     Right (S3.HeadObjectMemoryResponse (Just _)) -> do
-      object <- observeObjectStream store key
+      object <- getObjectHeaders store key
       pure (Just <$> object)
 
+getObjectHeaders :: S3 -> Text -> IO (Either S3Error ObjectInfo)
+getObjectHeaders store key = do
+  result <- try (runResourceT $ do
+    response <- AwsRun.pureAws
+      (s3Configuration store)
+      (s3Service store)
+      (s3Manager store)
+      (S3.getObject (s3Bucket store) key)
+    let metadata = S3.gorMetadata response
+        httpResponse = S3.gorResponse response
+        knownSize =
+          lookup "Content-Length" (HTTP.responseHeaders httpResponse)
+            >>= readMaybe . BS8.unpack
+        knownIdentity =
+          lookup "nuthatch-sha256" (S3.omUserMetadata metadata)
+            >>= either (const Nothing) Just . validateContentSha256
+    maybe (fail "missing object content length")
+      (\size -> pure ObjectInfo
+        { oiSize = size
+        , oiSha256 = knownIdentity
+        })
+      knownSize)
+    :: IO (Either SomeException ObjectInfo)
+  pure (either (Left . classifyException) Right result)
+
+observeObjectFully :: S3 -> ObjectKey -> IO (Either S3Error ObjectInfo)
+observeObjectFully store (ObjectKey key) = observeObjectStream store key
+
 deleteObject :: S3 -> ObjectKey -> IO (Either S3Error ())
 deleteObject store (ObjectKey key) = do
   result <- try (runMemory store (S3.DeleteObject key (s3Bucket store)))
@@ -259,30 +484,18 @@ observeObjectStream store key = do
         (s3Service store)
         (s3Manager store)
         (S3.getObject (s3Bucket store) key)
-      let metadata = S3.gorMetadata response
-          httpResponse = S3.gorResponse response
-          knownSize =
-            lookup "Content-Length" (HTTP.responseHeaders httpResponse)
-              >>= readMaybe . BS8.unpack
-          knownIdentity =
-            lookup "nuthatch-sha256" (S3.omUserMetadata metadata)
-              >>= either (const Nothing) Just . validateContentSha256
-      case (knownSize, knownIdentity) of
-        (Just size, Just identity) ->
-          pure ObjectInfo
-            { oiSize = size
-            , oiSha256 = Just identity
-            }
-        _ -> do
-          (size, context) <- C.runConduit $
-            HTTP.responseBody httpResponse
-              C..| CL.fold updateDigest (0, hashInit)
-          pure ObjectInfo
-            { oiSize = size
-            , oiSha256 = Just (contextIdentity context)
-            }
+      let httpResponse = S3.gorResponse response
+      (size, context) <- C.runConduit $
+        HTTP.responseBody httpResponse
+          C..| CL.fold updateDigest (0, hashInit)
+      pure ObjectInfo
+        { oiSize = size
+        , oiSha256 = Just (contextIdentity context)
+        }
     updateDigest (size, context) chunk =
-      (size + toInteger (BS8.length chunk), hashUpdate context chunk)
+      let nextSize = size + toInteger (BS8.length chunk)
+          nextContext = hashUpdate context chunk
+      in nextSize `seq` nextContext `seq` (nextSize, nextContext)
 
 contextIdentity :: Context SHA256 -> ContentSha256
 contextIdentity =
@@ -311,6 +524,15 @@ classifyException exception =
         Just (HTTP.InvalidUrlException _ _) -> S3ResponseError
         Nothing -> S3ResponseError
 
+trySynchronous :: IO a -> IO (Either SomeException a)
+trySynchronous action = do
+  result <- try action
+  case result of
+    Left exception
+      | Just (_ :: SomeAsyncException) <- fromException exception ->
+          throwIO exception
+    _ -> pure result
+
 lastMaybe :: [a] -> Maybe a
 lastMaybe [] = Nothing
 lastMaybe values = Just (last values)

src/Nuthatch/Streaming.hs

@@ -0,0 +1,124 @@
+module Nuthatch.Streaming
+  ( UploadStreamError (..)
+  , BandwidthError (..)
+  , multipartPartBytes
+  , maximumMultipartBytes
+  , consumeUploadParts
+  ) where
+
+import           Control.Exception (Exception, throwIO)
+import           Crypto.Hash (Context, SHA256, hashFinalize, hashInit, hashUpdate)
+import qualified Data.ByteArray as BA
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Base16 as B16
+import           Data.IORef
+import qualified Data.Text.Encoding as TE
+import           Nuthatch.Types
+import           Servant.Types.SourceT
+  ( SourceT
+  , foreach
+  )
+
+data UploadStreamError
+  = UploadStreamInvalid String
+  | UploadStreamSizeMismatch
+  | UploadStreamIdentityMismatch
+  deriving stock (Eq, Show)
+
+instance Exception UploadStreamError
+
+data BandwidthError
+  = BandwidthUnavailable
+  | BandwidthExceeded
+  deriving stock (Eq, Show)
+
+instance Exception BandwidthError
+
+multipartPartBytes :: Int
+multipartPartBytes = 5 * 1024 * 1024
+
+maximumMultipartBytes :: Integer
+maximumMultipartBytes = toInteger multipartPartBytes * 10000
+
+data UploadState = UploadState
+  { usPending     :: [BS.ByteString]
+  , usPendingSize :: Int
+  , usSize        :: Integer
+  , usHash        :: Context SHA256
+  }
+
+consumeUploadParts
+  :: Integer
+  -> ContentSha256
+  -> Int
+  -> Int
+  -> (BS.ByteString -> IO ())
+  -> SourceT IO BS.ByteString
+  -> (BS.ByteString -> IO ())
+  -> IO ()
+consumeUploadParts
+    expectedSize
+    expectedIdentity
+    pieceSize
+    partSize
+    inspect
+    body
+    emitPart = do
+  stateRef <- newIORef UploadState
+    { usPending = []
+    , usPendingSize = 0
+    , usSize = 0
+    , usHash = hashInit
+    }
+  foreach
+    (throwIO . UploadStreamInvalid)
+    (mapM_ (acceptPiece stateRef) . splitPieces)
+    body
+  state <- readIORef stateRef
+  if usSize state /= expectedSize
+    then throwIO UploadStreamSizeMismatch
+    else if digestIdentity (usHash state) /= expectedIdentity
+      then throwIO UploadStreamIdentityMismatch
+      else if usPendingSize state == 0
+        then pure ()
+        else emitPart (BS.concat (reverse (usPending state)))
+  where
+    splitPieces bytes
+      | BS.null bytes = []
+      | otherwise =
+          let (piece, rest) = BS.splitAt pieceSize bytes
+          in piece : splitPieces rest
+
+    acceptPiece stateRef piece = do
+      inspect piece
+      state <- readIORef stateRef
+      let nextSize = usSize state + toInteger (BS.length piece)
+          nextHash = hashUpdate (usHash state) piece
+      nextHash `seq` pure ()
+      if nextSize > expectedSize
+        then throwIO UploadStreamSizeMismatch
+        else fillPart stateRef state
+          { usPending = piece : usPending state
+          , usPendingSize = usPendingSize state + BS.length piece
+          , usSize = nextSize
+          , usHash = nextHash
+          }
+
+    fillPart stateRef state
+      | usPendingSize state < partSize = writeIORef stateRef state
+      | otherwise = do
+          let bytes = BS.concat (reverse (usPending state))
+              (part, remainder) = BS.splitAt partSize bytes
+          emitPart part
+          writeIORef stateRef state
+            { usPending = [remainder | not (BS.null remainder)]
+            , usPendingSize = BS.length remainder
+            }
+
+digestIdentity :: Context SHA256 -> ContentSha256
+digestIdentity =
+  ContentSha256
+    . TE.decodeUtf8
+    . B16.encode
+    . BA.convert
+    . hashFinalize

test/Spec.hs

@@ -1,5 +1,6 @@
 module Main (main) where
 
+import           Control.Exception (try)
 import           Control.Monad.IO.Class (liftIO)
 import           Control.Lens ((&), (?~))
 import           Crypto.JOSE (runJOSE)
@@ -12,6 +13,7 @@ import           Crypto.JOSE.JWK
   , genJWK
   )
 import           Crypto.JOSE.JWS (newJWSHeaderProtected)
+import           Crypto.Hash (Digest, SHA256, hash)
 import           Crypto.JWT
   ( Audience (..)
   , ClaimsSet
@@ -23,7 +25,9 @@ import           Crypto.JWT
   )
 import qualified Data.Aeson as A
 import qualified Data.Aeson.KeyMap as KM
+import qualified Data.ByteArray as BA
 import qualified Data.ByteString as BS
+import qualified Data.ByteString.Base16 as B16
 import qualified Data.ByteString.Lazy as BL
 import           Data.Either (isLeft, isRight)
 import           Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)
@@ -87,10 +91,21 @@ import           Nuthatch.Protocol
   )
 import           Nuthatch.Types
 import           Nuthatch.Session (mintSession)
+import           Nuthatch.Streaming
+  ( UploadStreamError (..)
+  , consumeUploadParts
+  , maximumMultipartBytes
+  , multipartPartBytes
+  )
 import           Nuthatch.Ui (browsePage, browserScript)
 import           Servant
 import           Servant.Server.Experimental.Auth (AuthHandler, mkAuthHandler)
-import           Servant.Types.SourceT (source)
+import           Servant.Types.SourceT
+  ( SourceT
+  , StepT (..)
+  , fromStepT
+  , source
+  )
 import           Test.Tasty (TestTree, defaultMain, testGroup)
 import           Test.Tasty.HUnit ((@?=), assertBool, testCase)
 import           Test.Tasty.Hedgehog (testProperty)
@@ -103,10 +118,61 @@ tests = testGroup "nuthatch-unit"
   [ pathTests
   , identityTests
   , protocolTests
+  , streamingTests
   , oidcTests
   , apiTests
   ]
 
+streamingTests :: TestTree
+streamingTests = testGroup "streaming"
+  [ testCase "upload bodies are inspected and emitted in bounded parts" $ do
+      inspected <- newIORef []
+      emitted <- newIORef []
+      consumeUploadParts
+        6
+        (strictSha256 "abcdef")
+        2
+        4
+        (\piece -> modifyIORef' inspected (<> [piece]))
+        (source ["abcdef"])
+        (\part -> modifyIORef' emitted (<> [part]))
+      readIORef inspected >>= (@?= ["ab", "cd", "ef"])
+      readIORef emitted >>= (@?= ["abcd", "ef"])
+  , testCase "multipart buffering has a fixed part-count bound" $
+      maximumMultipartBytes @?= toInteger multipartPartBytes * 10000
+  , testCase "upload rejects early end, surplus bytes, and identity mismatch" $ do
+      short <- uploadFailure 4 (strictSha256 "body") (source ["bod"])
+      short @?= UploadStreamSizeMismatch
+      long <- uploadFailure 4 (strictSha256 "body") (source ["body!"])
+      long @?= UploadStreamSizeMismatch
+      identity <- uploadFailure 4 (strictSha256 "else") (source ["body"])
+      identity @?= UploadStreamIdentityMismatch
+  , testCase "upload preserves source errors" $ do
+      failure <- uploadFailure
+        4
+        (strictSha256 "body")
+        (fromStepT (Error "request failed"))
+      failure @?= UploadStreamInvalid "request failed"
+  ]
+
+uploadFailure
+  :: Integer
+  -> ContentSha256
+  -> SourceT IO BS.ByteString
+  -> IO UploadStreamError
+uploadFailure size identity body = do
+  result <- try $
+    consumeUploadParts
+      size identity 2 4 (const (pure ())) body (const (pure ()))
+  case result of
+    Left failure -> pure failure
+    Right _ -> fail "expected upload stream rejection"
+
+strictSha256 :: BS.ByteString -> ContentSha256
+strictSha256 bytes =
+  ContentSha256 . TE.decodeUtf8 . B16.encode . BA.convert $
+    (hash bytes :: Digest SHA256)
+
 pathTests :: TestTree
 pathTests = testGroup "canonical paths"
   [ testProperty "valid paths parse and render without rewriting" propPathRoundTrip
@@ -604,7 +670,7 @@ typedTestServer calls =
             }))
   :<|> (\_ _ _ -> counted
         (addHeader "attachment; filename=\"file.txt\""
-          (addHeader "nosniff" (source ["body"]))))
+          (addHeader "nosniff" (addHeader 4 (source ["body"])))))
   :<|> (\_ _ _ operation _ _ _ _ -> counted UploadResponse
         { uploadResponseOperationId = operation
         , uploadResponseResult = MutationSucceeded

tests/kind/wp05-job.yaml.in

@@ -362,6 +362,374 @@ spec:
                 exit 1
               fi
 
+              if [ "${NUTHATCH_STREAMING_MODE:-}" = large ]; then
+                large_bytes="${NUTHATCH_STREAMING_LARGE_BYTES:?}"
+                printf 'Preparing %s-byte streaming object\n' "$large_bytes"
+                large_digest="$(
+                  dd if=/dev/zero bs=1048576 \
+                    count="$((large_bytes / 1048576))" status=none |
+                    sha256sum |
+                    cut -d' ' -f1
+                )"
+                printf 'Uploading constrained-memory streaming object\n'
+                status="$(
+                  dd if=/dev/zero bs=1048576 \
+                    count="$((large_bytes / 1048576))" status=none |
+                    curl --silent --show-error \
+                      --output /tmp/large-upload.json \
+                      --write-out '%{http_code}' \
+                      --cookie "$cookies" \
+                      --header 'Content-Type: application/octet-stream' \
+                      --header "X-Declared-Size: $large_bytes" \
+                      --header "X-Content-SHA256: $large_digest" \
+                      --header "X-CSRF-Token: $csrf" \
+                      --request POST \
+                      --upload-file - \
+                      "$base/upload?root=personal&path=streaming-large.bin&operation_id=70000000-0000-4000-8000-000000000001"
+                )"
+                printf 'Large upload status=%s\n' "$status"
+                test "$status" = 200
+                grep -F '"result":"succeeded"' /tmp/large-upload.json
+                printf 'Downloading constrained-memory streaming object\n'
+                download_digest="$(
+                  curl --fail --silent --show-error \
+                    --cookie "$cookies" \
+                    "$base/download?root=personal&path=streaming-large.bin" |
+                    sha256sum |
+                    cut -d' ' -f1
+                )"
+                printf 'Large download digest expected=%s actual=%s\n' \
+                  "$large_digest" "$download_digest"
+                test "$download_digest" = "$large_digest"
+
+                printf x >/tmp/tiny
+                tiny_digest="$(sha256sum /tmp/tiny | cut -d' ' -f1)"
+                for attempt in first replay; do
+                  status="$(
+                    curl --silent --show-error \
+                      --output "/tmp/prior-desired-$attempt.json" \
+                      --write-out '%{http_code}' \
+                      --cookie "$cookies" \
+                      --header 'Content-Type: application/octet-stream' \
+                      --header "X-Declared-Size: $large_bytes" \
+                      --header "X-Content-SHA256: $large_digest" \
+                      --header "X-CSRF-Token: $csrf" \
+                      --data-binary @/tmp/tiny \
+                      "$base/upload?root=personal&path=streaming-large.bin&operation_id=70000000-0000-4000-8000-000000000007"
+                  )"
+                  test "$status" = 200
+                  grep -F '"result":"succeeded"' \
+                    "/tmp/prior-desired-$attempt.json"
+                done
+                status="$(
+                  curl --silent --show-error \
+                    --output /tmp/size-mismatch.json \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header 'X-Declared-Size: 2' \
+                    --header "X-Content-SHA256: $tiny_digest" \
+                    --header "X-CSRF-Token: $csrf" \
+                    --data-binary @/tmp/tiny \
+                    "$base/upload?root=personal&path=size-mismatch.bin&operation_id=70000000-0000-4000-8000-000000000002"
+                )"
+                printf 'Size mismatch status=%s body=' "$status"
+                cat /tmp/size-mismatch.json
+                printf '\n'
+                test "$status" = 400
+                grep -F '"error":"size_mismatch"' /tmp/size-mismatch.json
+                status="$(
+                  curl --silent --show-error \
+                    --output /tmp/identity-mismatch.json \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header 'X-Declared-Size: 1' \
+                    --header "X-Content-SHA256: $(printf '%064d' 0)" \
+                    --header "X-CSRF-Token: $csrf" \
+                    --data-binary @/tmp/tiny \
+                    "$base/upload?root=personal&path=identity-mismatch.bin&operation_id=70000000-0000-4000-8000-000000000003"
+                )"
+                printf 'Identity mismatch status=%s body=' "$status"
+                cat /tmp/identity-mismatch.json
+                printf '\n'
+                test "$status" = 400
+                grep -F '"error":"content_identity_mismatch"' \
+                  /tmp/identity-mismatch.json
+                status="$(
+                  set +o pipefail
+                  dd if=/dev/zero bs=1048576 count=160 status=none |
+                    curl --silent --show-error \
+                      --max-time 5 \
+                      --limit-rate 64k \
+                      --output /tmp/too-large.json \
+                      --write-out '%{http_code}' \
+                      --cookie "$cookies" \
+                      --header 'Content-Type: application/octet-stream' \
+                      --header "X-Declared-Size: $NUTHATCH_STREAMING_OVER_LIMIT_BYTES" \
+                      --header "X-Content-SHA256: $tiny_digest" \
+                      --header "X-CSRF-Token: $csrf" \
+                      --request POST \
+                      --upload-file - \
+                      "$base/upload?root=personal&path=too-large.bin&operation_id=70000000-0000-4000-8000-000000000004"
+                )"
+                printf 'Over-limit status=%s body=' "$status"
+                cat /tmp/too-large.json
+                printf '\n'
+                test "$status" = 413
+                grep -F '"error":"object_too_large"' /tmp/too-large.json
+
+                if dd if=/dev/zero bs=1048576 \
+                  count="$((large_bytes / 1048576))" status=none |
+                  curl --silent --show-error \
+                    --max-time 1 \
+                    --limit-rate 64k \
+                    --cookie "$cookies" \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header "X-Declared-Size: $large_bytes" \
+                    --header "X-Content-SHA256: $large_digest" \
+                    --header "X-CSRF-Token: $csrf" \
+                    --request POST \
+                    --upload-file - \
+                    "$base/upload?root=personal&path=aborted-upload.bin&operation_id=70000000-0000-4000-8000-000000000005" \
+                    >/tmp/aborted-upload.json; then
+                  exit 1
+                fi
+                aborted_upload_status="$(
+                  curl --silent --output /dev/null \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    "$base/download?root=personal&path=aborted-upload.bin"
+                )"
+                printf 'Aborted upload lookup status=%s\n' \
+                  "$aborted_upload_status"
+                test "$aborted_upload_status" = 404
+                for retry in $(seq 1 20); do
+                  status="$(
+                    curl --silent --show-error \
+                      --output /tmp/aborted-upload-replay.json \
+                      --write-out '%{http_code}' \
+                      --cookie "$cookies" \
+                      --header 'Content-Type: application/octet-stream' \
+                      --header "X-Declared-Size: $large_bytes" \
+                      --header "X-Content-SHA256: $large_digest" \
+                      --header "X-CSRF-Token: $csrf" \
+                      --data-binary @/tmp/tiny \
+                      "$base/upload?root=personal&path=aborted-upload.bin&operation_id=70000000-0000-4000-8000-000000000005"
+                  )"
+                  test "$status" = 200
+                  if grep -F '"result":"failed"' \
+                    /tmp/aborted-upload-replay.json >/dev/null; then
+                    break
+                  fi
+                  sleep 0.25
+                done
+                grep -F '"result":"failed"' /tmp/aborted-upload-replay.json
+                if dd if=/dev/zero bs=1048576 \
+                  count="$((large_bytes / 1048576))" status=none |
+                  curl --silent --show-error \
+                    --max-time 1 \
+                    --limit-rate 64k \
+                    --cookie "$cookies" \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header "X-Declared-Size: $large_bytes" \
+                    --header "X-Content-SHA256: $(printf '%064d' 0)" \
+                    --header "X-CSRF-Token: $csrf" \
+                    --request POST \
+                    --upload-file - \
+                    "$base/upload?root=personal&path=streaming-large.bin&operation_id=70000000-0000-4000-8000-000000000006" \
+                    >/tmp/aborted-overwrite.json; then
+                  exit 1
+                fi
+                preserved_digest="$(
+                  curl --fail --silent --show-error \
+                    --cookie "$cookies" \
+                    "$base/download?root=personal&path=streaming-large.bin" |
+                    sha256sum |
+                    cut -d' ' -f1
+                )"
+                test "$preserved_digest" = "$large_digest"
+                if curl --silent --show-error \
+                  --max-time 1 \
+                  --limit-rate 64k \
+                  --cookie "$cookies" \
+                  "$base/download?root=personal&path=streaming-large.bin" \
+                  >/tmp/aborted-download; then
+                  exit 1
+                fi
+                aborted_bytes="$(wc -c </tmp/aborted-download | tr -d ' ')"
+                test "$aborted_bytes" -gt 0
+                test "$aborted_bytes" -lt "$large_bytes"
+                curl --fail --silent --show-error "$base/readyz" |
+                  grep -F '"ready":true'
+                printf 'Large streaming and abort probes passed\n'
+                exit 0
+              fi
+
+              if [ "${NUTHATCH_STREAMING_MODE:-}" = throttle ]; then
+                csrf_b="$(login "$NUTHATCH_OIDC_USER_B" b)"
+                dd if=/dev/zero of=/tmp/joint-upload bs=65536 \
+                  count=1 status=none
+                joint_digest="$(
+                  sha256sum /tmp/joint-upload | cut -d' ' -f1
+                )"
+                status="$(
+                  curl --silent --show-error \
+                    --output /tmp/joint-upload.json \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header 'X-Declared-Size: 65536' \
+                    --header "X-Content-SHA256: $joint_digest" \
+                    --header "X-CSRF-Token: $csrf" \
+                    --data-binary @/tmp/joint-upload \
+                    "$base/upload?root=personal&path=joint-throttle.bin&operation_id=70000000-0000-4000-8000-000000000013"
+                )"
+                test "$status" = 200
+                grep -F '"result":"succeeded"' /tmp/joint-upload.json
+                dd if=/dev/zero of=/tmp/throttle-upload bs=1048576 \
+                  count=1 status=none
+                upload_digest="$(
+                  sha256sum /tmp/throttle-upload | cut -d' ' -f1
+                )"
+                status="$(
+                  curl --silent --show-error \
+                    --output /tmp/throttle-upload.json \
+                    --write-out '%{http_code}' \
+                    --cookie /tmp/cookies-b \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header 'X-Declared-Size: 1048576' \
+                    --header "X-Content-SHA256: $upload_digest" \
+                    --header "X-CSRF-Token: $csrf_b" \
+                    --data-binary @/tmp/throttle-upload \
+                    "$base/upload?root=personal&path=throttled-upload.bin&operation_id=71000000-0000-4000-8000-000000000001"
+                )"
+                test "$status" = 429
+                grep -F '"error":"bandwidth_exceeded"' /tmp/throttle-upload.json
+                printf x >/tmp/tiny
+                test "$(
+                  curl --silent --output /tmp/prestart-upload.json \
+                    --write-out '%{http_code}' \
+                    --cookie /tmp/cookies-b \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header 'X-Declared-Size: 1' \
+                    --header "X-Content-SHA256: $upload_digest" \
+                    --header "X-CSRF-Token: $csrf_b" \
+                    --data-binary @/tmp/tiny \
+                    "$base/upload?root=personal&path=prestart.bin&operation_id=71000000-0000-4000-8000-000000000002"
+                )" = 429
+
+                curl --silent --show-error --cookie "$cookies" \
+                  "$base/download?root=personal&path=streaming-large.bin" \
+                  >/tmp/parallel-a &
+                first_pid=$!
+                curl --silent --show-error --cookie "$cookies" \
+                  "$base/download?root=personal&path=streaming-large.bin" \
+                  >/tmp/parallel-b &
+                second_pid=$!
+                first_ok=true
+                second_ok=true
+                wait "$first_pid" || first_ok=false
+                wait "$second_pid" || second_ok=false
+                test "$first_ok" = false
+                test "$second_ok" = false
+                parallel_bytes="$(
+                  wc -c /tmp/parallel-a /tmp/parallel-b |
+                    awk 'END {print $1}'
+                )"
+                printf 'Parallel downloads moved %s bytes with burst %s\n' \
+                  "$parallel_bytes" \
+                  "$NUTHATCH_STREAMING_THROTTLE_BURST_BYTES"
+                test "$parallel_bytes" -gt 0
+                joint_limit="$(
+                  expr "$NUTHATCH_STREAMING_THROTTLE_BURST_BYTES" - 65536 + 4096
+                )"
+                test "$parallel_bytes" -le "$joint_limit"
+                prestart_status="$(
+                  curl --silent --output /tmp/prestart-download.json \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    "$base/download?root=personal&path=streaming-large.bin"
+                )"
+                printf 'Post-exhaustion download status=%s\n' "$prestart_status"
+                test "$prestart_status" = 429
+                printf 'Shared incremental throttle probes passed\n'
+                exit 0
+              fi
+
+              if [ "${NUTHATCH_STREAMING_MODE:-}" = refill ]; then
+                dd if=/dev/zero of=/tmp/refill-upload bs=65536 \
+                  count=1 status=none
+                refill_digest="$(
+                  sha256sum /tmp/refill-upload | cut -d' ' -f1
+                )"
+                for suffix in initial refilled; do
+                  if [ "$suffix" = refilled ]; then
+                    sleep 2
+                    refill_operation=72000000-0000-4000-8000-000000000002
+                  else
+                    refill_operation=72000000-0000-4000-8000-000000000001
+                  fi
+                  status="$(
+                    curl --silent --show-error \
+                      --output "/tmp/refill-$suffix.json" \
+                      --write-out '%{http_code}' \
+                      --cookie "$cookies" \
+                      --header 'Content-Type: application/octet-stream' \
+                      --header 'X-Declared-Size: 65536' \
+                      --header "X-Content-SHA256: $refill_digest" \
+                      --header "X-CSRF-Token: $csrf" \
+                      --data-binary @/tmp/refill-upload \
+                      "$base/upload?root=personal&path=refill-$suffix.bin&operation_id=$refill_operation"
+                  )"
+                  test "$status" = 200
+                  grep -F '"result":"succeeded"' "/tmp/refill-$suffix.json"
+                  if [ "$suffix" = initial ]; then
+                    test "$(
+                      curl --silent --output /tmp/refill-exhausted.json \
+                        --write-out '%{http_code}' \
+                        --cookie "$cookies" \
+                        --header 'Content-Type: application/octet-stream' \
+                        --header 'X-Declared-Size: 65536' \
+                        --header "X-Content-SHA256: $refill_digest" \
+                        --header "X-CSRF-Token: $csrf" \
+                        --data-binary @/tmp/refill-upload \
+                        "$base/upload?root=personal&path=refill-exhausted.bin&operation_id=72000000-0000-4000-8000-000000000003"
+                    )" = 429
+                  fi
+                done
+                printf 'Token-bucket exhaustion and refill probes passed\n'
+                exit 0
+              fi
+
+              if [ "${NUTHATCH_STREAMING_MODE:-}" = unavailable ]; then
+                printf x >/tmp/tiny
+                tiny_digest="$(sha256sum /tmp/tiny | cut -d' ' -f1)"
+                test "$(
+                  curl --silent --output /tmp/unavailable-upload.json \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    --header 'Content-Type: application/octet-stream' \
+                    --header 'X-Declared-Size: 1' \
+                    --header "X-Content-SHA256: $tiny_digest" \
+                    --header "X-CSRF-Token: $csrf" \
+                    --data-binary @/tmp/tiny \
+                    "$base/upload?root=personal&path=unavailable.bin&operation_id=72000000-0000-4000-8000-000000000001"
+                )" = 503
+                grep -F '"error":"throttle_unavailable"' \
+                  /tmp/unavailable-upload.json
+                test "$(
+                  curl --silent --output /tmp/unavailable-download.json \
+                    --write-out '%{http_code}' \
+                    --cookie "$cookies" \
+                    "$base/download?root=personal&path=streaming-large.bin"
+                )" = 503
+                grep -F '"error":"throttle_unavailable"' \
+                  /tmp/unavailable-download.json
+                printf 'Unavailable coordination probes passed\n'
+                exit 0
+              fi
+
               exercise_root() {
                 root="$1"
                 suffix="$2"
@@ -563,6 +931,30 @@ spec:
                 configMapKeyRef:
                   name: @@PROFILE_NAME@@
                   key: NUTHATCH_OIDC_REDIRECT_URI
+            - name: NUTHATCH_STREAMING_MODE
+              valueFrom:
+                configMapKeyRef:
+                  name: @@PROFILE_NAME@@
+                  key: NUTHATCH_STREAMING_MODE
+                  optional: true
+            - name: NUTHATCH_STREAMING_LARGE_BYTES
+              valueFrom:
+                configMapKeyRef:
+                  name: @@PROFILE_NAME@@
+                  key: NUTHATCH_STREAMING_LARGE_BYTES
+                  optional: true
+            - name: NUTHATCH_STREAMING_OVER_LIMIT_BYTES
+              valueFrom:
+                configMapKeyRef:
+                  name: @@PROFILE_NAME@@
+                  key: NUTHATCH_STREAMING_OVER_LIMIT_BYTES
+                  optional: true
+            - name: NUTHATCH_STREAMING_THROTTLE_BURST_BYTES
+              valueFrom:
+                configMapKeyRef:
+                  name: @@PROFILE_NAME@@
+                  key: NUTHATCH_STREAMING_THROTTLE_BURST_BYTES
+                  optional: true
             - name: NUTHATCH_OIDC_CLIENT_SECRET
               valueFrom:
                 secretKeyRef:

tests/kind/wp07-run.sh

@@ -0,0 +1,217 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
+source "$ROOT/tests/kind/lib.sh"
+
+USER_KUBECONFIG_BEFORE="$(user_kubeconfig_digest)"
+SUCCEEDED=false
+OWNS_FIXTURE=false
+APP_IMAGE=nuthatch-wp07:local
+
+cleanup() {
+  local status=$?
+  trap - EXIT INT TERM
+  if [[ "${NUTHATCH_TESTBED_RETAIN:-0}" = 1 ]] && [[ "$SUCCEEDED" != true ]]; then
+    printf 'retained kubeconfig: %s\n' "$KUBECONFIG_PATH"
+    printf 'retained context: %s\n' "$CONTEXT"
+    exit "$status"
+  fi
+  if [[ "$OWNS_FIXTURE" = true ]]; then
+    destroy_fixture || status=1
+  fi
+  if [[ "$(user_kubeconfig_digest)" != "$USER_KUBECONFIG_BEFORE" ]]; then
+    printf 'user kubeconfig changed unexpectedly\n' >&2
+    status=1
+  fi
+  exit "$status"
+}
+
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+render_application() {
+  local lane="$1"
+  local profile_file="$2"
+  local rate="$3"
+  local burst="$4"
+  local pg_conn
+  local redis_url
+  local oidc_issuer
+  local oidc_client_id
+  local oidc_additional_audiences
+  local oidc_client_secret
+  local s3_endpoint
+  local s3_region
+  local s3_bucket
+  local s3_access_key
+  local s3_secret_key
+  pg_conn="$(sed -n 's/^NUTHATCH_PG_CONN=//p' "$profile_file")"
+  redis_url="$(sed -n 's/^NUTHATCH_REDIS_URL=//p' "$profile_file")"
+  oidc_issuer="$(sed -n 's/^NUTHATCH_OIDC_ISSUER=//p' "$profile_file")"
+  oidc_client_id="$(sed -n 's/^NUTHATCH_OIDC_CLIENT_ID=//p' "$profile_file")"
+  oidc_additional_audiences="$(
+    sed -n 's/^NUTHATCH_OIDC_ADDITIONAL_AUDIENCES=//p' "$profile_file"
+  )"
+  oidc_client_secret="$(<"$profile_file.oidc-secret")"
+  s3_endpoint="$(sed -n 's/^NUTHATCH_S3_ENDPOINT=//p' "$profile_file")"
+  s3_region="$(sed -n 's/^NUTHATCH_S3_REGION=//p' "$profile_file")"
+  s3_bucket="$(sed -n 's/^NUTHATCH_S3_BUCKET=//p' "$profile_file")"
+  s3_access_key="$(sed -n 's/^NUTHATCH_S3_ACCESS_KEY=//p' "$profile_file")"
+  s3_secret_key="$(sed -n 's/^NUTHATCH_S3_SECRET_KEY=//p' "$profile_file")"
+  sed \
+    -e "s|@@LANE@@|$lane|g" \
+    -e "s|@@APP_IMAGE@@|$APP_IMAGE|g" \
+    -e "s|@@PG_CONN@@|$pg_conn|g" \
+    -e "s|@@REDIS_URL@@|$redis_url|g" \
+    -e "s|@@OIDC_ISSUER@@|$oidc_issuer|g" \
+    -e "s|@@OIDC_CLIENT_ID@@|$oidc_client_id|g" \
+    -e "s|@@OIDC_ADDITIONAL_AUDIENCES@@|$oidc_additional_audiences|g" \
+    -e "s|@@OIDC_CLIENT_SECRET@@|$oidc_client_secret|g" \
+    -e "s|@@S3_ENDPOINT@@|$s3_endpoint|g" \
+    -e "s|@@S3_REGION@@|$s3_region|g" \
+    -e "s|@@S3_BUCKET@@|$s3_bucket|g" \
+    -e "s|@@S3_ACCESS_KEY@@|$s3_access_key|g" \
+    -e "s|@@S3_SECRET_KEY@@|$s3_secret_key|g" \
+    "$ROOT/deploy/kubernetes/nuthatch.yaml.in" > "$RUNTIME/wp07-app-$lane.yaml"
+  sed -i \
+    "/NUTHATCH_S3_PATH_STYLE/a\\  NUTHATCH_MAX_OBJECT_BYTES: \"201326592\"\\n  NUTHATCH_THROTTLE_BYTES_PER_SEC: \"$rate\"\\n  NUTHATCH_THROTTLE_BURST_BYTES: \"$burst\"" \
+    "$RUNTIME/wp07-app-$lane.yaml"
+  sed -i 's/  replicas: 1/  replicas: 2/' "$RUNTIME/wp07-app-$lane.yaml"
+  sed -i \
+    's/limits: {cpu: "2", memory: 512Mi}/limits: {cpu: "2", memory: 128Mi}/' \
+    "$RUNTIME/wp07-app-$lane.yaml"
+  sed -i \
+    '/          envFrom:/i\          env:\n            - {name: GHCRTS, value: "-N2 -A1m -M96m"}' \
+    "$RUNTIME/wp07-app-$lane.yaml"
+}
+
+cleanup_application() {
+  local lane="$1"
+  kube --namespace "$FIXTURE_NAMESPACE" delete \
+    "deployment/nuthatch-$lane" \
+    "service/nuthatch-$lane" \
+    "configmap/nuthatch-$lane" \
+    "secret/nuthatch-$lane" \
+    --ignore-not-found >/dev/null
+  rm -f "$RUNTIME/wp07-app-$lane.yaml"
+}
+
+set_streaming_profile() {
+  local profile="$1"
+  local mode="$2"
+  local throttle_burst=262144
+  local profile_file="$RUNTIME/$profile.env"
+  if [[ "$mode" = refill ]]; then
+    throttle_burst=65536
+  fi
+  sed -i '/^NUTHATCH_STREAMING_/d' "$profile_file"
+  {
+    printf 'NUTHATCH_STREAMING_MODE=%s\n' "$mode"
+    printf 'NUTHATCH_STREAMING_LARGE_BYTES=167772160\n'
+    printf 'NUTHATCH_STREAMING_OVER_LIMIT_BYTES=201326593\n'
+    printf 'NUTHATCH_STREAMING_THROTTLE_BURST_BYTES=%s\n' "$throttle_burst"
+  } >> "$profile_file"
+  kube create configmap "$profile" \
+    --namespace "$FIXTURE_NAMESPACE" \
+    --from-env-file="$profile_file" \
+    --dry-run=client \
+    --output yaml |
+    kube apply --filename - >/dev/null
+}
+
+run_streaming_job() {
+  local lane="$1"
+  local profile="$2"
+  local mode="$3"
+  local job="wp07-$mode-$lane"
+  set_streaming_profile "$profile" "$mode"
+  render_lane_job \
+    "$job" \
+    "$TESTBED/wp05-job.yaml.in" \
+    "$CURL_LOCAL_IMAGE" \
+    "$profile"
+  sed -i "s|@@LANE@@|$lane|g" "$RUNTIME/$job.yaml"
+  run_job_manifest "$job" "$RUNTIME/$job.yaml"
+}
+
+scale_coordination() {
+  local provider="$1"
+  local replicas="$2"
+  kube --namespace "$FIXTURE_NAMESPACE" scale \
+    "deployment/$provider" \
+    --replicas="$replicas" >/dev/null
+  if [[ "$replicas" = 0 ]]; then
+    kube --namespace "$FIXTURE_NAMESPACE" wait \
+      --for=delete \
+      "pod" \
+      --selector="app=$provider" \
+      --timeout=120s >/dev/null
+  else
+    kube --namespace "$FIXTURE_NAMESPACE" rollout status \
+      "deployment/$provider" \
+      --timeout=180s >/dev/null
+  fi
+}
+
+deploy_application() {
+  local lane="$1"
+  local profile="$2"
+  local rate="$3"
+  local burst="$4"
+  cleanup_application "$lane"
+  render_application "$lane" "$RUNTIME/$profile.env" "$rate" "$burst"
+  kube apply --filename "$RUNTIME/wp07-app-$lane.yaml" >/dev/null
+  kube --namespace "$FIXTURE_NAMESPACE" rollout status \
+    "deployment/nuthatch-$lane" --timeout=180s
+}
+
+acquire_fixture_lock
+OWNS_FIXTURE=true
+setup_fixture durable coordination oidc target
+source "$RUNTIME/fixture.env"
+"$NUTHATCH_REAL_NERDCTL" --namespace "$NERDCTL_NAMESPACE" build \
+  --file "$ROOT/Dockerfile" \
+  --tag "$APP_IMAGE" \
+  "$ROOT"
+load_local_image "$APP_IMAGE" "$NERDCTL_NAMESPACE"
+
+while IFS=$'\t' read -r lane coordination oidc target; do
+  if [[ -n "${NUTHATCH_WP07_LANE:-}" ]] &&
+     [[ "$lane" != "$NUTHATCH_WP07_LANE" ]]; then
+    continue
+  fi
+  profile="$(write_lane_profile "$lane" "$coordination" "$oidc" "$target")"
+  printf 'Running WP07 lane %s: coordination=%s oidc=%s target=%s\n' \
+    "$lane" "$coordination" "$oidc" "$target"
+
+  deploy_application "$lane" "$profile" 536870912 536870912
+  run_streaming_job "$lane" "$profile" large
+
+  cleanup_application "$lane"
+  scale_coordination "$coordination" 0
+  scale_coordination "$coordination" 1
+  deploy_application "$lane" "$profile" 1 262144
+  run_streaming_job "$lane" "$profile" throttle
+
+  deploy_application "$lane" "$profile" 65536 65536
+  run_streaming_job "$lane" "$profile" refill
+
+  deploy_application "$lane" "$profile" 536870912 536870912
+  scale_coordination "$coordination" 0
+  run_streaming_job "$lane" "$profile" unavailable
+  cleanup_application "$lane"
+  scale_coordination "$coordination" 1
+
+  cleanup_lane_state "$lane" "$coordination" "$oidc" "$target"
+  for mode in large throttle refill unavailable; do
+    cleanup_lane_resources "wp07-$mode-$lane" "$profile"
+  done
+  if [[ "${NUTHATCH_WP07_FIRST_LANE_ONLY:-0}" = 1 ]]; then
+    break
+  fi
+done < <(expand_fixture_matrix coordination oidc target)
+
+SUCCEEDED=true
+printf 'Nuthatch WP-07 streaming gate passed\n'