From 839f888ebe0eaea070fbb6009012aa3c3db119ce Mon Sep 17 00:00:00 2001 From: Michal Kozakiewicz Date: Thu, 30 Jul 2026 10:59:30 +0000 Subject: [PATCH 1/3] Download remote cache with parallel ranged GETs, and report transfer speed Restoring a cache used a single GetObject, i.e. one HTTP stream, which tends to be limited well below the available bandwidth. downloadObject now splits the object into byte ranges and fetches several at a time, emitting the chunks in order so tar sees an unchanged byte stream. Controlled by TASKRUNNER_S3_DOWNLOAD_CONCURRENCY (default 1) and TASKRUNNER_S3_DOWNLOAD_CHUNK_SIZE_MIB (default 8). The default of 1 keeps the old single-request behaviour, so this can be A/B tested on real CI before becoming the default. The first ranged request doubles as the existence check (HeadObject would not do: S3 answers HEAD with an empty body, so a missing object does not come back as _NoSuchKey) and its Content-Range gives the total size, so the remaining ranges are planned without an extra round trip. Prefetching starts from those response headers, so it overlaps with streaming the first chunk into tar. Ordering and bounded lookahead live in Control.Concurrent.Prefetch: a queue of Asyncs in item order, holding at most concurrency + 1 results in memory. Servers that mishandle Range are handled explicitly rather than silently truncating the archive - a non-206 response means we already have the whole object, and a 206 without a size falls back to one plain request. Also report size and speed for both directions as debug messages, addressing the long-standing TODO. Tests: DownloadTest runs downloadObject against a fake S3 (warp), so the ranged logic is covered without S3 credentials - byte-exact reassembly and request counts for multi-chunk, boundary, sliding-window, sub-chunk, concurrency-1, Range-ignoring, size-hiding and failing-chunk cases. The remote-cache-parallel-download golden test covers it end to end against a real S3. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 2 +- README.md | 2 + package.yaml | 1 + src/Control/Concurrent/Prefetch.hs | 73 ++++++++ src/RemoteCache.hs | 171 +++++++++++++++++-- src/Utils.hs | 25 +++ taskrunner.cabal | 6 + test/DownloadTest.hs | 193 ++++++++++++++++++++++ test/FakeS3.hs | 73 ++++++++ test/Spec.hs | 4 +- test/download-object.out | 9 + test/t/remote-cache-parallel-download.out | 8 + test/t/remote-cache-parallel-download.txt | 46 ++++++ 13 files changed, 599 insertions(+), 14 deletions(-) create mode 100644 src/Control/Concurrent/Prefetch.hs create mode 100644 test/DownloadTest.hs create mode 100644 test/FakeS3.hs create mode 100644 test/download-object.out create mode 100644 test/t/remote-cache-parallel-download.out create mode 100644 test/t/remote-cache-parallel-download.txt diff --git a/CLAUDE.md b/CLAUDE.md index ed025ae..21ba137 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -84,7 +84,7 @@ stack test --test-arguments --accept ## Notes - This project uses tasty-golden for snapshot/golden file testing - The test suite includes integration tests that verify taskrunner behavior -- **S3 Test Auto-Detection**: 15 tests require S3 credentials (marked with `# s3` directive in test files) +- **S3 Test Auto-Detection**: 16 tests require S3 credentials (marked with `# s3` directive in test files) - `stack test` automatically skips S3 tests if credentials are missing - To run S3 tests, set: `TASKRUNNER_TEST_S3_ENDPOINT`, `TASKRUNNER_TEST_S3_ACCESS_KEY`, `TASKRUNNER_TEST_S3_SECRET_KEY` - Use `SKIP_S3_TESTS=1` to explicitly skip S3 tests even when credentials are present diff --git a/README.md b/README.md index 22cb060..7a4d183 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,8 @@ To use it, first build using another system, and the run `taskrunner` with `TASK - `TASKRUNNER_DEBUG` - whether to output debug messages to toplevel output. Note that debug messages are always written to per-task logs, regardless of this setting. - `TASKRUNNER_LOG_INFO` - whether to output "info" messages to toplevel output. They are minimal messages, produced only when there's actually something to be done (including fetching from cache). +- `TASKRUNNER_S3_DOWNLOAD_CONCURRENCY` (default: `1`) - how many ranged `GET` requests to use in parallel when downloading a remote cache archive. A single stream is usually limited well below the available bandwidth, so raising this (e.g. to `8`) speeds up restoring large caches. `1` means a single plain request, as before. Transfer sizes and speeds are reported as debug messages. +- `TASKRUNNER_S3_DOWNLOAD_CHUNK_SIZE_MIB` (default: `8`) - how much a single ranged `GET` request asks for. At most `TASKRUNNER_S3_DOWNLOAD_CONCURRENCY + 1` chunks are held in memory at a time. - more... ## Possible features diff --git a/package.yaml b/package.yaml index 9c0f0f2..c694bac 100644 --- a/package.yaml +++ b/package.yaml @@ -31,6 +31,7 @@ dependencies: - unix - process - async +- stm - time - temporary - optparse-applicative diff --git a/src/Control/Concurrent/Prefetch.hs b/src/Control/Concurrent/Prefetch.hs new file mode 100644 index 0000000..0fc2cf5 --- /dev/null +++ b/src/Control/Concurrent/Prefetch.hs @@ -0,0 +1,73 @@ +-- | Fetch a list of items concurrently, but consume the results strictly in +-- order. Useful for turning a sequence of independent, latency-bound requests +-- into a stream without buffering everything in memory at once. +module Control.Concurrent.Prefetch + ( Prefetch + , startPrefetch + , cancelPrefetch + , nextPrefetch + ) where + +import Universum + +import Control.Concurrent.Async (Async, async, cancel, wait) +import Control.Concurrent.STM (TBQueue, newTBQueueIO, readTBQueue, writeTBQueue) +import Control.Exception (mask_) +import Data.List (delete) + +data Prefetch a = Prefetch + { queue :: TBQueue (Maybe (Async a)) + -- ^ Results in item order. 'Nothing' marks the end of the stream. + , producer :: Async () + , inFlight :: TVar [Async a] + -- ^ Fetches that have been started but not yet consumed, so that + -- 'cancelPrefetch' can stop them. Consumed fetches are removed, otherwise + -- we would keep every result alive until the whole stream is done. + } + +-- | Start fetching @items@ in the background, at most @concurrency + 1@ at a +-- time, and hand them out in order via 'nextPrefetch'. +-- +-- Memory use is bounded by the size of @concurrency + 1@ results, since a fetch +-- is only started once there is room for its result. +-- +-- Must be paired with 'cancelPrefetch' (via 'bracket' or similar), which is +-- also what makes exceptions safe: if a fetch fails, 'nextPrefetch' rethrows it +-- and 'cancelPrefetch' stops the remaining ones. +startPrefetch :: Int -> [i] -> (i -> IO a) -> IO (Prefetch a) +startPrefetch concurrency items fetch = do + queue <- newTBQueueIO (fromIntegral (max 1 concurrency)) + inFlight <- newTVarIO [] + producer <- async do + forM_ items \item -> do + -- Registering the fetch must not be interruptible, or a cancellation + -- landing in between would leave an unreachable thread running. + a <- mask_ do + a <- async (fetch item) + atomically $ modifyTVar' inFlight (a:) + pure a + -- Blocks while the consumer is behind, which is what bounds concurrency. + atomically $ writeTBQueue queue (Just a) + atomically $ writeTBQueue queue Nothing + pure Prefetch{queue, producer, inFlight} + +-- | Stop the producer and any outstanding fetches. Idempotent. +-- +-- Note that this makes any concurrent 'nextPrefetch' block forever, so only +-- call it once the consumer is done with the stream. +cancelPrefetch :: Prefetch a -> IO () +cancelPrefetch prefetch = do + -- Cancel the producer first, so that it cannot start anything new while we + -- are cancelling what is already in flight. + cancel prefetch.producer + readTVarIO prefetch.inFlight >>= mapM_ cancel + +-- | Next result in item order, or 'Nothing' once all items have been handed +-- out. Rethrows whatever the corresponding fetch threw. +nextPrefetch :: Prefetch a -> IO (Maybe a) +nextPrefetch prefetch = + atomically (readTBQueue prefetch.queue) >>= \case + Nothing -> + pure Nothing + Just a -> + Just <$> wait a `finally` atomically (modifyTVar' prefetch.inFlight (delete a)) diff --git a/src/RemoteCache.hs b/src/RemoteCache.hs index 48fa670..17bda60 100644 --- a/src/RemoteCache.hs +++ b/src/RemoteCache.hs @@ -3,10 +3,10 @@ module RemoteCache where import Universum -import Control.Monad.Trans.Resource (MonadResource) +import Control.Monad.Trans.Resource (MonadResource, ResourceT) import Amazonka.Env (newEnv, Env'(..), overrideService) import Amazonka.S3 (BucketName(..), ObjectKey(..), newGetObject, _NoSuchKey, StorageClass (StorageClass_REDUCED_REDUNDANCY)) -import Amazonka.S3.GetObject (GetObjectResponse(..)) +import Amazonka.S3.GetObject (GetObject(..), GetObjectResponse(..)) import qualified Data.ByteString as BS import Data.Conduit ((.|), ConduitT, bracketP, runConduitRes) import qualified Data.Conduit.Zstd as Zstd @@ -21,20 +21,22 @@ import System.Environment (lookupEnv) import Amazonka.Types ( Region(..), AccessKey(..), SecretKey(..), Service, s3AddressingStyle, S3AddressingStyle(..) ) import Types import System.Process (CreateProcess(..), cleanupProcess, createProcess_, StdStream (..), proc, waitForProcess) -import Conduit (sourceHandle, sinkHandle, foldMapC) +import Conduit (sourceHandle, sinkHandle, foldMapC, sinkList) import Network.URI (parseURI, URI (..), URIAuth(..)) import System.Directory (makeAbsolute, canonicalizePath) import System.FilePath (makeRelative) import qualified System.FilePath as FP -import Utils (bail, logDebug, logFileName, logInfo, withStderrPipe) +import Utils (bail, bytesfmt, logDebug, logFileName, logInfo, logWarn, timed, transferSummary, withStderrPipe) import qualified Amazonka as AWS import Control.Exception.Lens (handling) import System.Exit (ExitCode(..)) import qualified Data.Conduit as C import qualified Data.Conduit.Text as CT +import qualified Data.Text as Text import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Builder as TLB import Amazonka.S3.PutObject (newPutObject, PutObject(..)) +import Control.Concurrent.Prefetch (cancelPrefetch, nextPrefetch, startPrefetch) packTar :: MonadResource m => AppState -> Handle -> FilePath -> [FilePath] -> ConduitT () BS.ByteString m () @@ -89,6 +91,12 @@ data RemoteCacheSettings = RemoteCacheSettings , logsPrefix :: Text , logsViewUrl :: Text + + -- | How many ranged GET requests to use when downloading a cache archive. + -- 1 (the default) means a single plain request for the whole object. + , s3DownloadConcurrency :: Int + -- | How many bytes a single ranged GET request asks for. + , s3DownloadChunkSize :: Int } getRemoteCacheSettingsFromEnv :: MonadIO m => m RemoteCacheSettings @@ -101,8 +109,23 @@ getRemoteCacheSettingsFromEnv = liftIO do remoteCachePrefix <- maybe "taskrunner/" toText <$> lookupEnv "TASKRUNNER_REMOTE_CACHE_PREFIX" logsPrefix <- maybe (error "TASKRUNNER_LOGS_PREFIX not provided") toText <$> lookupEnv "TASKRUNNER_LOGS_PREFIX" logsViewUrl <- maybe (error "TASKRUNNER_LOGS_VIEW_URL not provided") toText <$> lookupEnv "TASKRUNNER_LOGS_VIEW_URL" + s3DownloadConcurrency <- lookupPositiveIntEnv "TASKRUNNER_S3_DOWNLOAD_CONCURRENCY" 1 + s3DownloadChunkSizeMiB <- lookupPositiveIntEnv "TASKRUNNER_S3_DOWNLOAD_CHUNK_SIZE_MIB" 8 + let s3DownloadChunkSize = s3DownloadChunkSizeMiB * 1024 * 1024 pure RemoteCacheSettings{..} +lookupPositiveIntEnv :: String -> Int -> IO Int +lookupPositiveIntEnv name defaultValue = + lookupEnv name >>= \case + Nothing -> + pure defaultValue + Just str -> + case readMaybe str of + Just value | value > 0 -> + pure value + _ -> + error $ toText name <> " must be a positive integer, got: " <> show str + parseEndpoint :: Text -> Maybe (Service -> Service) parseEndpoint "default-aws" = Just id parseEndpoint s = do @@ -114,7 +137,6 @@ parseEndpoint s = do . (\svc -> svc { s3AddressingStyle = S3AddressingStylePath }) -- TODO: --- - report speed, size etc. -- - integrate amazonka logging -- - handle errors saveCache @@ -145,13 +167,18 @@ saveCache appState settings relativeCacheRoot files archiveName = do logDebug appState $ "Uploading to s3://" <> bucket <> "/" <> objectKey - withStderrPipe appState \stderrHandle -> + packedBytes <- newIORef 0 + uploadedBytes <- newIORef 0 + + (_, elapsed) <- timed $ withStderrPipe appState \stderrHandle -> runConduitRes do let multipartUpload = (newCreateMultipartUpload (BucketName bucket) (ObjectKey objectKey) :: CreateMultipartUpload) { storageClass = Just StorageClass_REDUCED_REDUNDANCY } result <- packTar appState stderrHandle cacheRoot filesRelativeToCacheRoot + .| countBytes packedBytes .| Zstd.compress 3 + .| countBytes uploadedBytes .| streamUpload env Nothing multipartUpload case result of Left (_, err) -> @@ -160,6 +187,19 @@ saveCache appState settings relativeCacheRoot files archiveName = do liftIO $ logDebug appState "Upload success" pure () + packed <- readIORef packedBytes + uploaded <- readIORef uploadedBytes + -- Note the rate covers the whole pipeline (tar, zstd and the upload), not + -- just the network part. + logDebug appState $ "Packed and uploaded " <> transferSummary uploaded elapsed + <> ", compressed from " <> toText (bytesfmt "%.2f" packed) + +-- | Pass data through unchanged, accumulating the total number of bytes seen. +countBytes :: MonadIO m => IORef Int -> ConduitT BS.ByteString BS.ByteString m () +countBytes ref = C.awaitForever \chunk -> do + modifyIORef' ref (+ BS.length chunk) + C.yield chunk + data LogMode = NoLog | Log deriving (Eq, Show) restoreCache @@ -181,14 +221,121 @@ restoreCache appState settings cacheRoot archiveName logMode = do logDebug appState $ "Remote cache archive not found s3://" <> bucket <> "/" <> objectKey pure False - handling _NoSuchKey onNoSuchKey $ withStderrPipe appState \stderrHandle -> - runConduitRes do - response <- AWS.send env $ newGetObject (BucketName bucket) (ObjectKey objectKey) + onFound = when (logMode == Log) do - liftIO $ logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" + logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" + + handling _NoSuchKey onNoSuchKey $ withStderrPipe appState \stderrHandle -> do + downloadedBytes <- newIORef 0 + + (_, elapsed) <- timed $ runConduitRes $ + downloadObject appState settings env (BucketName bucket) (ObjectKey objectKey) onFound + .| countBytes downloadedBytes + .| unpackTar appState stderrHandle cacheRoot + + downloaded <- readIORef downloadedBytes + -- The size is that of the compressed archive, and the rate covers the whole + -- pipeline (the download, zstd and tar), not just the network part. + logDebug appState $ "Downloaded and unpacked " <> transferSummary downloaded elapsed + + pure True + +-- | Stream an S3 object, using several parallel ranged GET requests when +-- @s3DownloadConcurrency@ is above 1. A single stream tends to be limited well +-- below the available bandwidth, so fetching a few ranges at once is +-- noticeably faster for large archives. +-- +-- Chunks are emitted strictly in order, so downstream sees the same byte stream +-- either way. +downloadObject + :: AppState + -> RemoteCacheSettings + -> AWS.Env + -> BucketName + -> ObjectKey + -> IO () -- ^ Called once the object is known to exist + -> ConduitT () BS.ByteString (ResourceT IO) () +downloadObject appState settings env bucket key onFound + | settings.s3DownloadConcurrency <= 1 = do + response <- AWS.send env $ newGetObject bucket key + liftIO onFound response.body.body - .| unpackTar appState stderrHandle cacheRoot - pure True + | otherwise = do + -- The first request doubles as the existence check (so that _NoSuchKey is + -- still thrown from here) and tells us the total size via Content-Range, + -- which is what lets us plan the remaining ranges without a separate + -- HeadObject request. Note that HeadObject would not do: S3 answers HEAD + -- with an empty body, so a missing object does not come back as + -- _NoSuchKey there. + firstResponse <- AWS.send env $ rangedGetObject bucket key (0, fromIntegral chunkSize - 1) + liftIO onFound + + -- A 206 means the range was honoured and the body is only the first + -- chunk; anything else (a server ignoring Range, or an object smaller + -- than one chunk served whole) means we already have everything. + if firstResponse.httpStatus /= 206 then + firstResponse.body.body + else case parseContentRangeTotal =<< firstResponse.contentRange of + Nothing -> do + -- Partial response, but we cannot tell how much is left, so we cannot + -- safely stream this body and stop. Start over in a single request. + liftIO $ logWarn appState $ "Could not determine object size from Content-Range: " + <> show firstResponse.contentRange <> ", downloading in a single request" + response <- AWS.send env $ newGetObject bucket key + response.body.body + Just total -> do + liftIO $ logDebug appState $ "Object size: " <> toText (bytesfmt "%.2f" total) + <> ", downloading with concurrency " <> show settings.s3DownloadConcurrency + case chunkRanges chunkSize (fromIntegral chunkSize) total of + [] -> + -- Object fits in a single chunk, which we already have. + firstResponse.body.body + remainingRanges -> + -- Start prefetching the rest right away, so it overlaps with + -- streaming the first chunk downstream. + bracketP + (startPrefetch settings.s3DownloadConcurrency remainingRanges + (fetchRange env bucket key)) + cancelPrefetch + \prefetch -> do + firstResponse.body.body + let go = liftIO (nextPrefetch prefetch) >>= \case + Nothing -> pure () + Just chunk -> C.yield chunk >> go + go + where + chunkSize = max 1 settings.s3DownloadChunkSize + +-- | Download a single byte range of an object into memory. +fetchRange :: AWS.Env -> BucketName -> ObjectKey -> (Integer, Integer) -> IO BS.ByteString +fetchRange env bucket key range' = + AWS.runResourceT do + response <- AWS.send env $ rangedGetObject bucket key range' + BS.concat <$> C.runConduit (response.body.body .| sinkList) + +-- | A GET request for an inclusive byte range, as in the HTTP @Range@ header. +rangedGetObject :: BucketName -> ObjectKey -> (Integer, Integer) -> GetObject +rangedGetObject bucket key (start, end) = + (newGetObject bucket key) + { range = Just $ "bytes=" <> show start <> "-" <> show end } + +-- | Split @[start, total)@ into consecutive inclusive ranges of at most +-- @chunkSize@ bytes each. +chunkRanges :: Int -> Integer -> Integer -> [(Integer, Integer)] +chunkRanges chunkSize start total + | start >= total = [] + | otherwise = + (start, min (start + size) total - 1) : chunkRanges chunkSize (start + size) total + where + size = fromIntegral (max 1 chunkSize) + +-- | Total object size from a @Content-Range@ header value, e.g. the 52428800 in +-- @bytes 0-8388607/52428800@. 'Nothing' if the size is unknown (@*@) or the +-- header is malformed. +parseContentRangeTotal :: Text -> Maybe Integer +parseContentRangeTotal header = do + let total = Text.drop 1 $ Text.dropWhile (/= '/') header + readMaybe (toString total) getLatestBuildHash :: AppState diff --git a/src/Utils.hs b/src/Utils.hs index 30a67eb..ecf9828 100644 --- a/src/Utils.hs +++ b/src/Utils.hs @@ -17,6 +17,7 @@ import GHC.IO.Handle (hIsClosed) import System.FilePath (()) import Control.Concurrent.Async (async, wait) import System.Timeout (timeout) +import GHC.Clock (getMonotonicTime) outputLine :: AppState -> Handle -> ByteString -> ByteString -> IO () outputLine appState toplevelOutput streamName line = do @@ -90,6 +91,30 @@ bytesfmt formatter bs = printf (formatter <> " %s") bytesSuffixes = ["B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"] bytesSuffix = bytesSuffixes !! i +-- | Run an action, also returning how long it took, in seconds. +timed :: MonadIO m => m a -> m (a, Double) +timed action = do + start <- liftIO getMonotonicTime + result <- action + end <- liftIO getMonotonicTime + pure (result, end - start) + +-- | Describe a transfer of @bytes@ bytes taking @seconds@ seconds. +-- +-- >>> transferSummary 12345678 1.5 +-- "11.77 MiB in 1.50s (7.85 MiB/s)" +transferSummary :: Int -> Double -> Text +transferSummary bytes seconds = + toText (bytesfmt "%.2f" bytes) <> " in " <> toText (printf "%.2fs" seconds :: String) <> rate + where + rate + -- Below that the rate is mostly measurement noise, and dividing by a very + -- small elapsed time gives an absurd number. + | seconds >= 0.01 = + " (" <> toText (bytesfmt "%.2f" (round (fromIntegral bytes / seconds) :: Int)) <> "/s)" + | otherwise = + "" + -- | Create a per-subprocess stderr pipe that prefixes output with the job name. -- The pipe is fully drained before returning. withStderrPipe :: AppState -> (Handle -> IO a) -> IO a diff --git a/taskrunner.cabal b/taskrunner.cabal index 710f969..fc19d9d 100644 --- a/taskrunner.cabal +++ b/taskrunner.cabal @@ -28,6 +28,7 @@ library App CliArgs CommitStatus + Control.Concurrent.Prefetch Control.Monad.EarlyReturn RemoteCache SnapshotCliArgs @@ -107,6 +108,7 @@ library , process , random , resourcet + , stm , tar , tar-conduit , temporary @@ -193,6 +195,7 @@ executable taskrunner , process , random , resourcet + , stm , tar , tar-conduit , taskrunner @@ -209,7 +212,9 @@ test-suite taskrunner-test type: exitcode-stdio-1.0 main-is: Spec.hs other-modules: + DownloadTest FakeGithubApi + FakeS3 Paths_taskrunner autogen-modules: Paths_taskrunner @@ -282,6 +287,7 @@ test-suite taskrunner-test , process , random , resourcet + , stm , tar , tar-conduit , taskrunner diff --git a/test/DownloadTest.hs b/test/DownloadTest.hs new file mode 100644 index 0000000..c378130 --- /dev/null +++ b/test/DownloadTest.hs @@ -0,0 +1,193 @@ +-- | Tests for 'downloadObject', which splits a download into parallel ranged +-- GET requests. These run against 'FakeS3' and need no S3 credentials. +module DownloadTest (tests) where + +import Universum + +import App (getSettings) +import Conduit (runResourceT, sinkList) +import qualified Amazonka as AWS +import Amazonka.Auth (fromKeys) +import Amazonka.Env (newEnv, Env'(..), overrideService) +import Amazonka.S3 (BucketName(..), ObjectKey(..)) +import Amazonka.Types (AccessKey(..), Region(..), SecretKey(..)) +import qualified Data.ByteString as BS +import Data.Conduit ((.|)) +import qualified Data.Conduit as C +import qualified Data.Text as Text +import RemoteCache (RemoteCacheSettings(..), downloadObject, parseEndpoint) +import System.IO (IOMode(..)) +import Test.Tasty (TestTree) +import Test.Tasty.Golden (goldenVsStringDiff) +import Types + +import FakeS3 (Behaviour(..), RequestLog, withFakeS3) + +mib :: Int +mib = 1024 * 1024 + +tests :: TestTree +tests = + goldenVsStringDiff + "download-object" + (\ref new -> ["diff", "-u", ref, new]) + "test/download-object.out" + (encodeUtf8 . unlines <$> mapM runCase cases) + +data Case = Case + { name :: Text + , behaviour :: Behaviour + , objectSize :: Int + , concurrency :: Int + , chunkSize :: Int + , expected :: Int -> Either Text ByteString -> [Maybe ByteString] -> Text + -- ^ Given the object size, the download result and the requests the server + -- saw, describe the outcome. + } + +cases :: [Case] +cases = + [ Case + { name = "several chunks" + , behaviour = HonourRange, objectSize = 3000083, concurrency = 4, chunkSize = mib + , expected = expectObject + } + , Case + { name = "exact multiple of chunk size" + -- Must not ask for an extra, empty range past the end. + , behaviour = HonourRange, objectSize = 2 * mib, concurrency = 4, chunkSize = mib + , expected = expectObject + } + , Case + { name = "one byte over a chunk boundary" + , behaviour = HonourRange, objectSize = mib + 1, concurrency = 4, chunkSize = mib + , expected = expectObject + } + , Case + { name = "more chunks than concurrency" + -- The prefetch window has to slide, rather than deadlock or reorder. + , behaviour = HonourRange, objectSize = 10 * mib, concurrency = 2, chunkSize = mib + , expected = expectObject + } + , Case + { name = "smaller than one chunk" + , behaviour = HonourRange, objectSize = 100, concurrency = 4, chunkSize = mib + , expected = expectObject + } + , Case + { name = "concurrency 1 does not use ranges" + , behaviour = HonourRange, objectSize = 3000083, concurrency = 1, chunkSize = mib + , expected = \size result requests -> + expectObject size result requests <> ", range headers: " <> show requests + } + , Case + { name = "server ignores Range" + -- We must notice we got the whole object, and not truncate it. + , behaviour = IgnoreRange, objectSize = 3000083, concurrency = 4, chunkSize = mib + , expected = expectObject + } + , Case + { name = "server hides the object size" + -- Nothing to base the ranges on, so it should start over unranged. + , behaviour = UnknownTotal, objectSize = 3000083, concurrency = 4, chunkSize = mib + , expected = expectObject + } + , Case + { name = "a chunk fails" + -- Must fail loudly rather than hand a truncated archive to tar. The + -- request count is not checked, since amazonka retries and the remaining + -- chunks may or may not have been started. + , behaviour = FailAtOffset mib, objectSize = 3000083, concurrency = 4, chunkSize = mib + , expected = \_ result _ -> case result of + Left _ -> "failed, as expected" + Right bytes -> "SUCCEEDED UNEXPECTEDLY with " <> show (BS.length bytes) <> " bytes" + } + ] + +-- | Check we got the object back byte for byte, and report how many requests it +-- took (which is the point of the whole exercise). +expectObject :: Int -> Either Text ByteString -> [Maybe ByteString] -> Text +expectObject size result requests = + case result of + Left err -> + "FAILED: " <> err + Right bytes + | bytes == payload size -> + "ok in " <> show (length requests) <> " request(s)" + | BS.length bytes /= size -> + "WRONG LENGTH: got " <> show (BS.length bytes) <> ", expected " <> show size + | otherwise -> + "WRONG CONTENT (right length) - chunks reordered or overlapping?" + +runCase :: Case -> IO Text +runCase testCase = do + let object = payload testCase.objectSize + appState <- mkAppState + withFakeS3 testCase.behaviour object \requestLog port -> do + env <- mkEnv port + result <- tryDownload appState (mkSettings port testCase) env + requests <- readIORef requestLog + pure $ testCase.name <> ": " <> testCase.expected testCase.objectSize result requests + +tryDownload + :: AppState -> RemoteCacheSettings -> AWS.Env -> IO (Either Text ByteString) +tryDownload appState settings env = do + result <- try @IO @SomeException $ runResourceT $ C.runConduit $ + downloadObject appState settings env (BucketName "bucket") (ObjectKey "obj") pass + .| (BS.concat <$> sinkList) + pure $ first (Text.unwords . Text.words . Text.take 200 . show) result + +-- | Deterministic filler that zstd cannot squash, so that test objects actually +-- stay big enough to span several chunks. +payload :: Int -> ByteString +payload size = BS.pack $ take size $ cycle + [fromIntegral (i * 7 + i `div` 251) | i <- [0 :: Int .. 4095]] + +mkEnv :: Int -> IO AWS.Env +mkEnv port = do + let endpoint = "http://localhost:" <> show port + endpointFn = fromMaybe (error "invalid endpoint") $ parseEndpoint endpoint + newEnv (pure . fromKeys (AccessKey "key") (SecretKey "secret")) + -- Silent logger: one case deliberately provokes a server error, and it + -- should not look like the test itself went wrong. + <&> (\env -> env { region = Region' "eu-central-1", logger = \_ _ -> pass }) + . overrideService endpointFn + +mkSettings :: Int -> Case -> RemoteCacheSettings +mkSettings port testCase = RemoteCacheSettings + { s3Endpoint = "http://localhost:" <> show port + , awsRegion = "eu-central-1" + , awsAccessKey = "key" + , awsSecretKey = "secret" + , remoteCacheBucket = "bucket" + , remoteCachePrefix = "" + , logsPrefix = "" + , logsViewUrl = "" + , s3DownloadConcurrency = testCase.concurrency + , s3DownloadChunkSize = testCase.chunkSize + } + +-- | Just enough 'AppState' for the logging that 'downloadObject' does. Log +-- output is discarded, so that it cannot end up in the golden output. +mkAppState :: IO AppState +mkAppState = do + settings <- getSettings + hashToSaveRef <- newIORef Nothing + snapshotArgsRef <- newIORef Nothing + skipped <- newIORef False + quietBuffer <- newIORef [] + githubClient <- newIORef Nothing + devNull <- openFile "/dev/null" WriteMode + pure AppState + { settings + , jobName = "download-test" + , buildId = "test" + , isToplevel = True + , hashToSaveRef + , snapshotArgsRef + , skipped + , toplevelStderr = devNull + , logOutput = devNull + , quietBuffer + , githubClient + } diff --git a/test/FakeS3.hs b/test/FakeS3.hs new file mode 100644 index 0000000..f15f9e8 --- /dev/null +++ b/test/FakeS3.hs @@ -0,0 +1,73 @@ +-- | A minimal stand-in for S3 that serves a single object, with just enough +-- behaviour to exercise ranged downloads (including servers that handle Range +-- badly). +module FakeS3 + ( Behaviour(..) + , RequestLog + , withFakeS3 + ) where + +import Universum + +import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as B8 +import qualified Data.ByteString.Lazy as LBS +import Data.List (lookup) +import qualified Network.HTTP.Types as HTTP +import qualified Network.Wai as Wai +import qualified Network.Wai.Handler.Warp as Warp + +data Behaviour + = HonourRange + -- ^ Like S3: answer a Range request with 206 and a Content-Range header. + | IgnoreRange + -- ^ Ignore Range and answer with the whole object and 200. + | UnknownTotal + -- ^ Answer with 206, but without disclosing the object size. + | FailAtOffset Int + -- ^ Fail requests for the range starting at the given offset. + deriving (Eq, Show) + +-- | Range header of every request the server received, in arrival order. +type RequestLog = IORef [Maybe ByteString] + +-- | Serve @object@ on a free port, and pass that port to the given action. +withFakeS3 :: Behaviour -> ByteString -> (RequestLog -> Int -> IO a) -> IO a +withFakeS3 behaviour object action = do + requestLog <- newIORef [] + Warp.testWithApplication (pure (app behaviour object requestLog)) (action requestLog) + +app :: Behaviour -> ByteString -> RequestLog -> Wai.Application +app behaviour object requestLog request respond = do + let m_range = lookup HTTP.hRange (Wai.requestHeaders request) + modifyIORef' requestLog (<> [m_range]) + let whole = respond $ Wai.responseLBS HTTP.status200 + [("Content-Length", show (BS.length object))] (LBS.fromStrict object) + case (behaviour, m_range >>= parseRange) of + (IgnoreRange, _) -> + whole + (_, Nothing) -> + whole + (FailAtOffset offset, Just (start, _)) | start == offset -> + respond $ Wai.responseLBS HTTP.status500 [] + "InternalError" + (_, Just (start, end)) -> do + let lastByte = min end (BS.length object - 1) + body = BS.take (lastByte - start + 1) $ BS.drop start object + total = case behaviour of + UnknownTotal -> "*" + _ -> show (BS.length object) + respond $ Wai.responseLBS HTTP.status206 + [ ("Content-Range", "bytes " <> show start <> "-" <> show lastByte <> "/" <> total) + , ("Content-Length", show (BS.length body)) + ] (LBS.fromStrict body) + +-- | Parse an inclusive @bytes=first-last@ range. Other forms are not used by +-- taskrunner, so they are treated as no range at all. +parseRange :: ByteString -> Maybe (Int, Int) +parseRange header = do + rest <- B8.stripPrefix "bytes=" header + let (startStr, rest') = B8.break (== '-') rest + start <- readMaybe (B8.unpack startStr) + end <- readMaybe (B8.unpack (B8.drop 1 rest')) + pure (start, end) diff --git a/test/Spec.hs b/test/Spec.hs index b83cdb6..1bcd923 100644 --- a/test/Spec.hs +++ b/test/Spec.hs @@ -28,6 +28,7 @@ import Amazonka.S3.ListObjectsV2 (ListObjectsV2Response(..)) import Amazonka.S3.Types.ObjectIdentifier (newObjectIdentifier) import Amazonka.S3.Types.Object (Object(..)) import qualified FakeGithubApi +import qualified DownloadTest main :: IO () main = defaultMain =<< goldenTests @@ -80,7 +81,8 @@ goldenTests = do System.IO.putStrLn $ "Running " <> show runningTests <> "/" <> show totalTests <> " tests" pure $ Tasty.withResource (FakeGithubApi.start fakeGithubPort) FakeGithubApi.stop \fakeGithubServer -> - testGroup "tests" + testGroup "tests" $ + DownloadTest.tests : [ goldenVsStringDiff (takeBaseName inputFile) -- test name (\ref new -> ["diff", "-u", ref, new]) diff --git a/test/download-object.out b/test/download-object.out new file mode 100644 index 0000000..e3c9700 --- /dev/null +++ b/test/download-object.out @@ -0,0 +1,9 @@ +several chunks: ok in 3 request(s) +exact multiple of chunk size: ok in 2 request(s) +one byte over a chunk boundary: ok in 2 request(s) +more chunks than concurrency: ok in 10 request(s) +smaller than one chunk: ok in 1 request(s) +concurrency 1 does not use ranges: ok in 1 request(s), range headers: [Nothing] +server ignores Range: ok in 1 request(s) +server hides the object size: ok in 2 request(s) +a chunk fails: failed, as expected diff --git a/test/t/remote-cache-parallel-download.out b/test/t/remote-cache-parallel-download.out new file mode 100644 index 0000000..334ccaf --- /dev/null +++ b/test/t/remote-cache-parallel-download.out @@ -0,0 +1,8 @@ +-- output: +[mytask] info | Inputs changed, running task +[mytask] stdout | Expensive computation +[mytask] info | success +2c05d8e8fa2f226c78c1c415a46d1d40be1fb01d output.bin +[mytask] info | Found remote cache mytask-393d94bb843a0f5f4911e8152e2736a29216f394.tar.zst, restoring +[mytask] info | Restored from remote cache +2c05d8e8fa2f226c78c1c415a46d1d40be1fb01d output.bin diff --git a/test/t/remote-cache-parallel-download.txt b/test/t/remote-cache-parallel-download.txt new file mode 100644 index 0000000..8f6941f --- /dev/null +++ b/test/t/remote-cache-parallel-download.txt @@ -0,0 +1,46 @@ +# no toplevel +# s3 + +export TASKRUNNER_SAVE_REMOTE_CACHE=1 + +# Download the archive using several ranged GET requests. Chunk size is small so +# that a few megabytes are enough to span multiple chunks. +export TASKRUNNER_S3_DOWNLOAD_CONCURRENCY=4 +export TASKRUNNER_S3_DOWNLOAD_CHUNK_SIZE_MIB=1 + +mkdir a + +( + cd a + # Deterministic, but incompressible - so that the archive actually stays big + # enough to be downloaded in several chunks. + head -c 3000000 /dev/zero | openssl enc -aes-256-ctr -pass pass:taskrunner -nosalt 2>/dev/null > input.bin + git init -q + git add input.bin + git commit -qm "Add input.bin" +) + +cp -r a b + +go() { + # Note: we want separate workdir for separate repos + TASKRUNNER_STATE_DIRECTORY="$(pwd)" taskrunner -n mytask bash -e -c ' + snapshot input.bin --outputs output.bin + echo "Expensive computation" + cp input.bin output.bin + ' +} + +( + cd a + go + sha1sum output.bin +) + +( + cd b + # Should download the cache instead of running the task, and the restored + # output must be byte-identical. + go + sha1sum output.bin +) From a72c6cdf3af3d2c6110c2c4d3c0059fdd6802405 Mon Sep 17 00:00:00 2001 From: Michal Kozakiewicz Date: Thu, 30 Jul 2026 12:06:56 +0000 Subject: [PATCH 2/3] Fix spurious zstd error on remote cache miss Folding the initial GetObject into the unpack pipeline broke every cache miss: conduit initialises sinks before pulling from the source, so unpackTar had already spawned "tar -x --zstd" by the time the request threw _NoSuchKey. The pipeline then unwound, tar's stdin was closed empty, and zstd reported "unexpected end of file" on the task's stderr - which showed up in 10 of the S3 golden tests. downloadObject becomes startDownload: it performs the initial request and *returns* a source, so restoreCache can sequence request, then the "Found remote cache" message, then unpacking - the order the code had before. That also removes the need for the onFound callback. DownloadTest gains a case for exactly this: with a missing object, startDownload must report NoSuchKey without the downstream sink having been started. Verified it discriminates - the old fused pipeline shape reports "downstream started: True". Also drop two redundant imports that CI warned about. Tested against a local MinIO with the same env CI uses: all 65 tests pass, including remote-cache-parallel-download, whose golden file was previously hand-written and is now confirmed. Co-Authored-By: Claude Opus 5 (1M context) --- src/RemoteCache.hs | 60 +++++++++++++++++++++++----------------- test/DownloadTest.hs | 55 ++++++++++++++++++++++++++---------- test/FakeS3.hs | 5 ++++ test/download-object.out | 1 + 4 files changed, 81 insertions(+), 40 deletions(-) diff --git a/src/RemoteCache.hs b/src/RemoteCache.hs index 17bda60..639f079 100644 --- a/src/RemoteCache.hs +++ b/src/RemoteCache.hs @@ -3,7 +3,7 @@ module RemoteCache where import Universum -import Control.Monad.Trans.Resource (MonadResource, ResourceT) +import Control.Monad.Trans.Resource (MonadResource, ResourceT, runResourceT) import Amazonka.Env (newEnv, Env'(..), overrideService) import Amazonka.S3 (BucketName(..), ObjectKey(..), newGetObject, _NoSuchKey, StorageClass (StorageClass_REDUCED_REDUNDANCY)) import Amazonka.S3.GetObject (GetObject(..), GetObjectResponse(..)) @@ -221,17 +221,20 @@ restoreCache appState settings cacheRoot archiveName logMode = do logDebug appState $ "Remote cache archive not found s3://" <> bucket <> "/" <> objectKey pure False - onFound = - when (logMode == Log) do - logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" - handling _NoSuchKey onNoSuchKey $ withStderrPipe appState \stderrHandle -> do downloadedBytes <- newIORef 0 - (_, elapsed) <- timed $ runConduitRes $ - downloadObject appState settings env (BucketName bucket) (ObjectKey objectKey) onFound - .| countBytes downloadedBytes - .| unpackTar appState stderrHandle cacheRoot + (_, elapsed) <- timed $ runResourceT do + source <- startDownload appState settings env (BucketName bucket) (ObjectKey objectKey) + + -- Only now that the archive is known to exist: say so, and start unpacking. + when (logMode == Log) do + logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" + + C.runConduit $ + source + .| countBytes downloadedBytes + .| unpackTar appState stderrHandle cacheRoot downloaded <- readIORef downloadedBytes -- The size is that of the compressed archive, and the rate covers the whole @@ -240,26 +243,32 @@ restoreCache appState settings cacheRoot archiveName logMode = do pure True --- | Stream an S3 object, using several parallel ranged GET requests when --- @s3DownloadConcurrency@ is above 1. A single stream tends to be limited well --- below the available bandwidth, so fetching a few ranges at once is --- noticeably faster for large archives. +-- | Make the initial request for an S3 object, and return a source streaming its +-- contents. Uses several parallel ranged GET requests when +-- @s3DownloadConcurrency@ is above 1: a single stream tends to be limited well +-- below the available bandwidth, so fetching a few ranges at once is noticeably +-- faster for large archives. -- -- Chunks are emitted strictly in order, so downstream sees the same byte stream -- either way. -downloadObject +-- +-- Note the first request deliberately happens before the returned source is +-- consumed, so that a missing object is reported (as '_NoSuchKey') before the +-- caller starts anything else. Conduit initialises sinks before pulling from the +-- source, so folding this into the pipeline would mean 'unpackTar' had already +-- spawned tar by the time we found out, which then complains about its empty +-- input on every cache miss. +startDownload :: AppState -> RemoteCacheSettings -> AWS.Env -> BucketName -> ObjectKey - -> IO () -- ^ Called once the object is known to exist - -> ConduitT () BS.ByteString (ResourceT IO) () -downloadObject appState settings env bucket key onFound + -> ResourceT IO (ConduitT () BS.ByteString (ResourceT IO) ()) +startDownload appState settings env bucket key | settings.s3DownloadConcurrency <= 1 = do response <- AWS.send env $ newGetObject bucket key - liftIO onFound - response.body.body + pure response.body.body | otherwise = do -- The first request doubles as the existence check (so that _NoSuchKey is -- still thrown from here) and tells us the total size via Content-Range, @@ -268,32 +277,31 @@ downloadObject appState settings env bucket key onFound -- with an empty body, so a missing object does not come back as -- _NoSuchKey there. firstResponse <- AWS.send env $ rangedGetObject bucket key (0, fromIntegral chunkSize - 1) - liftIO onFound -- A 206 means the range was honoured and the body is only the first -- chunk; anything else (a server ignoring Range, or an object smaller -- than one chunk served whole) means we already have everything. if firstResponse.httpStatus /= 206 then - firstResponse.body.body + pure firstResponse.body.body else case parseContentRangeTotal =<< firstResponse.contentRange of Nothing -> do -- Partial response, but we cannot tell how much is left, so we cannot -- safely stream this body and stop. Start over in a single request. - liftIO $ logWarn appState $ "Could not determine object size from Content-Range: " + logWarn appState $ "Could not determine object size from Content-Range: " <> show firstResponse.contentRange <> ", downloading in a single request" response <- AWS.send env $ newGetObject bucket key - response.body.body + pure response.body.body Just total -> do - liftIO $ logDebug appState $ "Object size: " <> toText (bytesfmt "%.2f" total) + logDebug appState $ "Object size: " <> toText (bytesfmt "%.2f" total) <> ", downloading with concurrency " <> show settings.s3DownloadConcurrency case chunkRanges chunkSize (fromIntegral chunkSize) total of [] -> -- Object fits in a single chunk, which we already have. - firstResponse.body.body + pure firstResponse.body.body remainingRanges -> -- Start prefetching the rest right away, so it overlaps with -- streaming the first chunk downstream. - bracketP + pure $ bracketP (startPrefetch settings.s3DownloadConcurrency remainingRanges (fetchRange env bucket key)) cancelPrefetch diff --git a/test/DownloadTest.hs b/test/DownloadTest.hs index c378130..c1f9111 100644 --- a/test/DownloadTest.hs +++ b/test/DownloadTest.hs @@ -9,19 +9,19 @@ import Conduit (runResourceT, sinkList) import qualified Amazonka as AWS import Amazonka.Auth (fromKeys) import Amazonka.Env (newEnv, Env'(..), overrideService) -import Amazonka.S3 (BucketName(..), ObjectKey(..)) +import Amazonka.S3 (BucketName(..), ObjectKey(..), _NoSuchKey) import Amazonka.Types (AccessKey(..), Region(..), SecretKey(..)) +import Control.Exception.Lens (handling) import qualified Data.ByteString as BS -import Data.Conduit ((.|)) +import Data.Conduit ((.|), bracketP) import qualified Data.Conduit as C import qualified Data.Text as Text -import RemoteCache (RemoteCacheSettings(..), downloadObject, parseEndpoint) -import System.IO (IOMode(..)) +import RemoteCache (RemoteCacheSettings(..), parseEndpoint, startDownload) import Test.Tasty (TestTree) import Test.Tasty.Golden (goldenVsStringDiff) import Types -import FakeS3 (Behaviour(..), RequestLog, withFakeS3) +import FakeS3 (Behaviour(..), withFakeS3) mib :: Int mib = 1024 * 1024 @@ -32,7 +32,10 @@ tests = "download-object" (\ref new -> ["diff", "-u", ref, new]) "test/download-object.out" - (encodeUtf8 . unlines <$> mapM runCase cases) + do + results <- mapM runCase cases + missing <- missingObjectCase + pure $ encodeUtf8 $ unlines $ results <> [missing] data Case = Case { name :: Text @@ -125,18 +128,42 @@ runCase testCase = do appState <- mkAppState withFakeS3 testCase.behaviour object \requestLog port -> do env <- mkEnv port - result <- tryDownload appState (mkSettings port testCase) env + result <- tryDownload appState (mkSettings port testCase.concurrency testCase.chunkSize) env requests <- readIORef requestLog pure $ testCase.name <> ": " <> testCase.expected testCase.objectSize result requests tryDownload :: AppState -> RemoteCacheSettings -> AWS.Env -> IO (Either Text ByteString) tryDownload appState settings env = do - result <- try @IO @SomeException $ runResourceT $ C.runConduit $ - downloadObject appState settings env (BucketName "bucket") (ObjectKey "obj") pass - .| (BS.concat <$> sinkList) + result <- try @IO @SomeException $ runResourceT do + source <- startDownload appState settings env (BucketName "bucket") (ObjectKey "obj") + C.runConduit $ source .| (BS.concat <$> sinkList) pure $ first (Text.unwords . Text.words . Text.take 200 . show) result +-- | A missing object has to be reported before anything downstream is started +-- up. 'restoreCache' relies on that: conduit initialises sinks before pulling +-- from the source, so if the initial request were part of the pipeline, tar +-- would already be running by the time the cache miss surfaced - and would +-- report a confusing error about its empty input on every cache miss. +missingObjectCase :: IO Text +missingObjectCase = do + appState <- mkAppState + withFakeS3 MissingObject "" \_ port -> do + env <- mkEnv port + let settings = mkSettings port 4 mib + sinkStarted <- newIORef False + outcome <- handling _NoSuchKey (\_ -> pure "reported as NoSuchKey") do + runResourceT do + source <- startDownload appState settings env (BucketName "bucket") (ObjectKey "obj") + C.runConduit $ source .| recordStartup sinkStarted + pure "NOT REPORTED" + started <- readIORef sinkStarted + pure $ "missing object: " <> outcome <> ", downstream started: " <> show started + where + -- Stands in for unpackTar: it is the bracketP allocation that spawns tar. + recordStartup ref = + bracketP (writeIORef ref True) (\() -> pass) \() -> C.awaitForever \_ -> pass + -- | Deterministic filler that zstd cannot squash, so that test objects actually -- stay big enough to span several chunks. payload :: Int -> ByteString @@ -153,8 +180,8 @@ mkEnv port = do <&> (\env -> env { region = Region' "eu-central-1", logger = \_ _ -> pass }) . overrideService endpointFn -mkSettings :: Int -> Case -> RemoteCacheSettings -mkSettings port testCase = RemoteCacheSettings +mkSettings :: Int -> Int -> Int -> RemoteCacheSettings +mkSettings port concurrency chunkSize = RemoteCacheSettings { s3Endpoint = "http://localhost:" <> show port , awsRegion = "eu-central-1" , awsAccessKey = "key" @@ -163,8 +190,8 @@ mkSettings port testCase = RemoteCacheSettings , remoteCachePrefix = "" , logsPrefix = "" , logsViewUrl = "" - , s3DownloadConcurrency = testCase.concurrency - , s3DownloadChunkSize = testCase.chunkSize + , s3DownloadConcurrency = concurrency + , s3DownloadChunkSize = chunkSize } -- | Just enough 'AppState' for the logging that 'downloadObject' does. Log diff --git a/test/FakeS3.hs b/test/FakeS3.hs index f15f9e8..634d463 100644 --- a/test/FakeS3.hs +++ b/test/FakeS3.hs @@ -26,6 +26,8 @@ data Behaviour -- ^ Answer with 206, but without disclosing the object size. | FailAtOffset Int -- ^ Fail requests for the range starting at the given offset. + | MissingObject + -- ^ Answer everything the way S3 reports an object that is not there. deriving (Eq, Show) -- | Range header of every request the server received, in arrival order. @@ -44,6 +46,9 @@ app behaviour object requestLog request respond = do let whole = respond $ Wai.responseLBS HTTP.status200 [("Content-Length", show (BS.length object))] (LBS.fromStrict object) case (behaviour, m_range >>= parseRange) of + (MissingObject, _) -> + respond $ Wai.responseLBS HTTP.status404 [] + "NoSuchKeyThe specified key does not exist." (IgnoreRange, _) -> whole (_, Nothing) -> diff --git a/test/download-object.out b/test/download-object.out index e3c9700..62b394d 100644 --- a/test/download-object.out +++ b/test/download-object.out @@ -7,3 +7,4 @@ concurrency 1 does not use ranges: ok in 1 request(s), range headers: [Nothing] server ignores Range: ok in 1 request(s) server hides the object size: ok in 2 request(s) a chunk fails: failed, as expected +missing object: reported as NoSuchKey, downstream started: False From 0007939cf2607bf86b28d42889694c413b20c074 Mon Sep 17 00:00:00 2001 From: Michal Kozakiewicz Date: Thu, 30 Jul 2026 12:52:55 +0000 Subject: [PATCH 3/3] v0.18.0.11 --- package.yaml | 2 +- taskrunner.cabal | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/package.yaml b/package.yaml index c694bac..17b1868 100644 --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: taskrunner -version: 0.18.0.10 +version: 0.18.0.11 github: "githubuser/taskrunner" license: BSD-3-Clause author: "Author name here" diff --git a/taskrunner.cabal b/taskrunner.cabal index fc19d9d..4966159 100644 --- a/taskrunner.cabal +++ b/taskrunner.cabal @@ -5,7 +5,7 @@ cabal-version: 2.2 -- see: https://github.com/sol/hpack name: taskrunner -version: 0.18.0.10 +version: 0.18.0.11 description: Please see the README on GitHub at homepage: https://github.com/githubuser/taskrunner#readme bug-reports: https://github.com/githubuser/taskrunner/issues