ms2k
15eb0038d2
Use descriptive test and executable names so implementation surfaces do not depend on journal organization. Consolidate the unreleased migration history into the authoritative initial schema because no deployed database requires an upgrade path.
667 lines
25 KiB
Haskell
667 lines
25 KiB
Haskell
module Main (main) where
|
|
|
|
import Control.Concurrent
|
|
( MVar
|
|
, newEmptyMVar
|
|
, putMVar
|
|
, takeMVar
|
|
, threadDelay
|
|
)
|
|
import Control.Concurrent.Async (poll, wait, withAsync)
|
|
import Control.Exception (SomeException, try)
|
|
import Control.Monad (forM, forM_, unless, when)
|
|
import Crypto.Hash (Digest, SHA256, hashlazy)
|
|
import qualified Data.ByteArray as BA
|
|
import qualified Data.ByteString.Base16 as B16
|
|
import qualified Data.ByteString as BS
|
|
import qualified Data.ByteString.Lazy as BL
|
|
import Data.Maybe (isNothing)
|
|
import Data.Text (Text)
|
|
import qualified Data.Text as T
|
|
import qualified Data.Text.Encoding as TE
|
|
import Database.PostgreSQL.Simple (Only (..))
|
|
import GHC.Clock (getMonotonicTimeNSec)
|
|
import Nuthatch.App
|
|
import Nuthatch.Config
|
|
import Nuthatch.Db
|
|
import qualified Nuthatch.Db as Db
|
|
import Nuthatch.Protocol
|
|
import Nuthatch.Redis
|
|
import Nuthatch.S3
|
|
import Nuthatch.Types
|
|
import Servant.Types.SourceT (SourceT (..), StepT (..), source)
|
|
import System.Environment (getArgs)
|
|
import System.IO (BufferMode (LineBuffering), hSetBuffering, stdout)
|
|
|
|
main :: IO ()
|
|
main = do
|
|
hSetBuffering stdout LineBuffering
|
|
cfg <- loadConfig
|
|
mode <- getArgs
|
|
case mode of
|
|
["pg-fence"] -> verifyPostgresUnavailable cfg
|
|
_ ->
|
|
withDb cfg $ \database -> do
|
|
migrate database "sql" cfg
|
|
withRedis cfg $ \redis -> do
|
|
store <- newS3 cfg
|
|
case mode of
|
|
["fence"] -> verifyStoreFence cfg database redis store
|
|
["restored"] -> verifyStoreRestored store
|
|
["store-prepare"] -> prepareStoreFence database store
|
|
["pg-prepare"] -> preparePostgresTransition cfg database >> spin
|
|
["pg-restored"] -> verifyPostgresRestored database
|
|
["full"] -> verifyConformance cfg database store
|
|
_ -> fail "unexpected conformance mode"
|
|
|
|
verifyConformance :: Config -> Db -> S3 -> IO ()
|
|
verifyConformance cfg database store = do
|
|
admissionScenarios cfg database
|
|
effectScenarios cfg database store
|
|
targetContract cfg database store
|
|
checkedInvariant "TypeOK"
|
|
putStrLn "Protocol conformance passed"
|
|
|
|
admissionScenarios :: Config -> Db -> IO ()
|
|
admissionScenarios cfg database = do
|
|
resetAuthority database 100
|
|
first <- admitPut cfg database "first" "first.bin" 10 identityA
|
|
replay <- admitPut cfg database "first" "first.bin" 10 identityA
|
|
conflict <- admitPut cfg database "first" "first.bin" 11 identityB
|
|
assert "FirstRequest" (isPrepared first)
|
|
assertEqual "ReplayRequest" MutationPending (replayResult replay)
|
|
assert "ConflictRequest" (isConflict conflict)
|
|
assertAuthority database 1 10 0
|
|
coveredAction "FirstRequest"
|
|
coveredAction "ReplayRequest"
|
|
coveredAction "ConflictRequest"
|
|
checkedInvariant "RecordIdentityConsistent"
|
|
checkedInvariant "FingerprintBindsRequest"
|
|
checkedInvariant "ObservationConsistent"
|
|
|
|
resetAuthority database 10
|
|
locked <- admitPut cfg database "locked-a" "locked.bin" 8 identityA
|
|
rejected <- admitPut cfg database "locked-b" "locked.bin" 8 identityA
|
|
quota <- admitPut cfg database "quota" "quota.bin" 8 identityA
|
|
assert "same-key rejection" (isPrepared locked && isRejected rejected)
|
|
assert "positive-delta quota rejection" (isRejected quota)
|
|
assertAuthority database 1 8 0
|
|
checkedInvariant "ReservationsConsistent"
|
|
checkedInvariant "LocksConsistent"
|
|
checkedInvariant "QuotaReservationBound"
|
|
|
|
resetAuthority database 100
|
|
let firstUser = UserId "personal-a"
|
|
secondUser = UserId "personal-b"
|
|
configurePoolLimit database (personalPool firstUser) 100
|
|
configurePoolLimit database (personalPool secondUser) 100
|
|
firstPersonal <- admitPersonal cfg database firstUser "personal-a"
|
|
secondPersonal <- admitPersonal cfg database secondUser "personal-b"
|
|
assert "personal admissions use disjoint pools"
|
|
(isPrepared firstPersonal && isPrepared secondPersonal)
|
|
personalRows <- Db.query_ database
|
|
"SELECT count(DISTINCT pool_id), count(DISTINCT object_key) \
|
|
\FROM mutations WHERE user_id IN ('personal-a', 'personal-b')"
|
|
:: IO [(Int, Int)]
|
|
assertEqual "personal pool and key separation" [(2, 2)] personalRows
|
|
checkedInvariant "PersonalRequestsUseOwnPool"
|
|
checkedInvariant "PersonalPoolsDisjoint"
|
|
|
|
effectScenarios :: Config -> Db -> S3 -> IO ()
|
|
effectScenarios cfg database store = do
|
|
resetScenario database store "applied.bin"
|
|
applied <- prepared =<< admitPut cfg database "applied" "applied.bin" 9 identityA
|
|
dispatch database applied
|
|
bounded cfg "put applied" $
|
|
expectRight =<< putObject store (mrObjectKey applied) 9 (Just identityA) "desired-a"
|
|
settle database applied
|
|
finalize database applied (ObjectPresent 9 (Just identityA)) MutationSucceeded
|
|
assertAuthority database 0 0 9
|
|
beforeReplay <- mutationSnapshot database "applied"
|
|
terminalReplay <- admitPut cfg database "applied" "applied.bin" 9 identityA
|
|
afterReplay <- mutationSnapshot database "applied"
|
|
assertEqual "terminal replay result" MutationSucceeded
|
|
(replayResult terminalReplay)
|
|
assertEqual "terminal replay preserves record" beforeReplay afterReplay
|
|
coveredAction "ExternalEffectApplied"
|
|
checkedInvariant "ActiveObjectConsistent"
|
|
checkedInvariant "ExternalStateConsistent"
|
|
checkedInvariant "ObjectStateCovered"
|
|
checkedInvariant "ObjectQuotaBound"
|
|
checkedInvariant "AccountingAtQuiescence"
|
|
checkedInvariant "RecordsNeverForgotten"
|
|
checkedInvariant "RequestBindingsStable"
|
|
checkedInvariant "TerminalRecordsStable"
|
|
|
|
resetScenario database store "unapplied.bin"
|
|
unapplied <- prepared =<< admitPut cfg database "unapplied" "unapplied.bin" 9 identityA
|
|
dispatch database unapplied
|
|
settle database unapplied
|
|
finalize database unapplied ObjectAbsent MutationFailed
|
|
assertAuthority database 0 0 0
|
|
coveredAction "ExternalEffectNotApplied"
|
|
|
|
resetScenario database store "body-loss.bin"
|
|
lost <- prepared =<< admitPut cfg database "body-loss" "body-loss.bin" 9 identityA
|
|
idleSettled <- withTransaction database $ \connection -> do
|
|
applying <- markApplying connection lost
|
|
settled <- markSettledFromIdle connection lost
|
|
pure (applying && settled)
|
|
assert "LoseUploadBody/SettleLostUpload" idleSettled
|
|
finalize database lost ObjectAbsent MutationFailed
|
|
assertAuthority database 0 0 0
|
|
coveredAction "LoseUploadBody"
|
|
coveredAction "SettleLostUpload"
|
|
checkedInvariant "UploadBodyConsistent"
|
|
checkedInvariant "UnfinishedEventuallyTerminal"
|
|
checkedInvariant "LostUploadEventuallyTerminal"
|
|
|
|
resetScenario database store "replacement.bin"
|
|
seedObject database store "replacement.bin" "prior-one"
|
|
replacement <- prepared
|
|
=<< admitPut cfg database "replacement" "replacement.bin" 9 identityB
|
|
dispatch database replacement
|
|
expectRight
|
|
=<< putObject store (mrObjectKey replacement) 9 (Just identityB) "after-two"
|
|
settle database replacement
|
|
finalize database replacement
|
|
(ObjectPresent 9 (Just identityB))
|
|
MutationSucceeded
|
|
assertAuthority database 0 0 9
|
|
|
|
resetScenario database store "delete-present.bin"
|
|
seedObject database store "delete-present.bin" "prior-one"
|
|
presentDelete <- prepared
|
|
=<< admitDelete cfg database "delete-present" "delete-present.bin"
|
|
dispatch database presentDelete
|
|
bounded cfg "delete present" $
|
|
expectRight =<< deleteObject store (mrObjectKey presentDelete)
|
|
settle database presentDelete
|
|
finalize database presentDelete ObjectAbsent MutationSucceeded
|
|
assertAuthority database 0 0 0
|
|
|
|
resetScenario database store "delete-absent.bin"
|
|
absentDelete <- prepared
|
|
=<< admitDelete cfg database "delete-absent" "delete-absent.bin"
|
|
dispatch database absentDelete
|
|
expectRight =<< deleteObject store (mrObjectKey absentDelete)
|
|
settle database absentDelete
|
|
finalize database absentDelete ObjectAbsent MutationSucceeded
|
|
assertAuthority database 0 0 0
|
|
|
|
resetScenario database store "unavailable.bin"
|
|
unavailable <- prepared
|
|
=<< admitPut cfg database "unavailable" "unavailable.bin" 9 identityA
|
|
dispatch database unavailable
|
|
settle database unavailable
|
|
result <- withTransaction database $ \connection ->
|
|
finalizeFromObservation connection unavailable
|
|
(ObjectUnavailable "injected")
|
|
assertEqual "observation unavailable stays active" Nothing result
|
|
assertAuthority database 1 9 0
|
|
|
|
targetContract :: Config -> Db -> S3 -> IO ()
|
|
targetContract cfg database store = do
|
|
resetScenario database store "atomic.bin"
|
|
let key = ObjectKey "shared/atomic.bin"
|
|
prior = BL.replicate (1024 * 1024) 65
|
|
desired = BL.replicate (8 * 1024 * 1024) 66
|
|
priorIdentity = contentIdentity prior
|
|
desiredIdentity = contentIdentity desired
|
|
expectRight
|
|
=<< putObject store key (toInteger (BL.length prior)) (Just priorIdentity) prior
|
|
uploadStarted <- newEmptyMVar
|
|
let replacement =
|
|
bounded cfg "atomic replacement" $
|
|
expectRight
|
|
=<< putObjectStream
|
|
store
|
|
key
|
|
(toInteger (BL.length desired))
|
|
(Just desiredIdentity)
|
|
(64 * 1024)
|
|
(const (pure ()))
|
|
(pacedSource uploadStarted desired)
|
|
withAsync replacement $ \writer -> do
|
|
takeMVar uploadStarted
|
|
overlap <- forM [1 .. 12 :: Int] $ \_ -> do
|
|
writing <- poll writer
|
|
observed <- expectRight =<< getObjectBytes store key
|
|
assert "atomic single-key replacement"
|
|
(observed == prior || observed == desired)
|
|
threadDelay 50000
|
|
pure (isNothing writing)
|
|
assert "replacement observations overlap an active put" (or overlap)
|
|
wait writer
|
|
stable <- expectRight =<< getObjectBytes store key
|
|
assertEqual "replacement settled to desired" desired stable
|
|
stableForSettlementBound cfg $ do
|
|
settled <- expectRight =<< getObjectBytes store key
|
|
assertEqual "no mutation after put settlement" desired settled
|
|
listed <- expectRight =<< listObjects store "shared/atomic.bin" Nothing Nothing 1
|
|
assert "ordinary list observes replacement"
|
|
(map fst (lpObjects listed) == ["shared/atomic.bin"])
|
|
bounded cfg "settled delete" $ expectRight =<< deleteObject store key
|
|
stableForSettlementBound cfg $ do
|
|
absent <- expectRight =<< headObject store key
|
|
assertEqual "no mutation after delete settlement" Nothing absent
|
|
checkedInvariant "SettledEffectsDoNotMutate"
|
|
|
|
verifyStoreFence :: Config -> Db -> Redis -> S3 -> IO ()
|
|
verifyStoreFence cfg database redis store = do
|
|
let env = newEnv cfg database redis store
|
|
user = SessionUser "protocol" "fence" protocolUser (Just "Protocol")
|
|
expiredDelete <- prepared
|
|
=<< admitDelete
|
|
cfg database "expired-delete-store-down" "expired-delete-store-down.bin"
|
|
expiredDeleteSettled <- withTransaction database $ \connection -> do
|
|
applying <- markApplying connection expiredDelete
|
|
settled <- markSettledFromIdle connection expiredDelete
|
|
pure (applying && settled)
|
|
assert "expired delete fixture settles" expiredDeleteSettled
|
|
finalize database expiredDelete ObjectAbsent MutationSucceeded
|
|
_ <- Db.execute database
|
|
"UPDATE mutations \
|
|
\SET updated_at = now() - (? * interval '1 second') \
|
|
\WHERE operation_id = 'expired-delete-store-down'"
|
|
(Only (cfgIdempotencyWindowSec cfg + 1))
|
|
beforeExpiredDelete <-
|
|
mutationSnapshot database "expired-delete-store-down"
|
|
expiredDeleteResult <- runDelete
|
|
env user Shared (RelativePath "expired-delete-store-down.bin")
|
|
(OperationId "expired-delete-store-down")
|
|
afterExpiredDelete <-
|
|
mutationSnapshot database "expired-delete-store-down"
|
|
assert "store-down expired delete epoch rejected" (isLeft expiredDeleteResult)
|
|
assertEqual "store-down does not start an expired delete epoch"
|
|
beforeExpiredDelete
|
|
afterExpiredDelete
|
|
|
|
freshDeleteResult <- runDelete
|
|
env user Shared (RelativePath "delete-store-down.bin")
|
|
(OperationId "delete-store-down")
|
|
assert "store-down delete rejected" (isLeft freshDeleteResult)
|
|
freshDeleteRows <- Db.query database
|
|
"SELECT count(*) FROM mutations WHERE operation_id = 'delete-store-down'"
|
|
()
|
|
:: IO [Only Int]
|
|
assertEqual "store-down delete creates no protocol record"
|
|
[Only 0]
|
|
freshDeleteRows
|
|
|
|
resetAuthority database 100
|
|
expired <- prepared
|
|
=<< admitPut cfg database "expired-store-down" "expired-store-down.bin" 9 identityA
|
|
expiredSettled <- withTransaction database $ \connection -> do
|
|
applying <- markApplying connection expired
|
|
settled <- markSettledFromIdle connection expired
|
|
pure (applying && settled)
|
|
assert "expired epoch fixture settles" expiredSettled
|
|
finalize database expired ObjectAbsent MutationFailed
|
|
_ <- Db.execute database
|
|
"UPDATE mutations \
|
|
\SET updated_at = now() - (? * interval '1 second') \
|
|
\WHERE operation_id = 'expired-store-down'"
|
|
(Only (cfgIdempotencyWindowSec cfg + 1))
|
|
beforeExpired <- mutationSnapshot database "expired-store-down"
|
|
expiredResult <- runPut
|
|
env
|
|
user
|
|
Shared
|
|
(RelativePath "expired-store-down.bin")
|
|
(OperationId "expired-store-down")
|
|
9
|
|
identityA
|
|
(source ["desired-a"])
|
|
afterExpired <- mutationSnapshot database "expired-store-down"
|
|
assert "store-down expired epoch rejected" (isLeft expiredResult)
|
|
assertEqual "store-down does not start an expired replacement epoch"
|
|
beforeExpired
|
|
afterExpired
|
|
|
|
active <- admitPut
|
|
cfg database "active-store-down" "active-store-down.bin" 9 identityA
|
|
assert "store-down active epoch fixture prepared" (isPrepared active)
|
|
beforeActive <- mutationSnapshot database "active-store-down"
|
|
activeReplay <- runPut
|
|
env user Shared (RelativePath "active-store-down.bin")
|
|
(OperationId "active-store-down") 9 identityA (source ["desired-a"])
|
|
activeConflict <- runPut
|
|
env user Shared (RelativePath "active-store-down.bin")
|
|
(OperationId "active-store-down") 9 identityB (source ["after-two"])
|
|
afterActive <- mutationSnapshot database "active-store-down"
|
|
assertEqual "store-down active replay" (Right MutationPending) activeReplay
|
|
assertEqual "store-down active conflict" (Right MutationConflict) activeConflict
|
|
assertEqual "store-down replay preserves active record" beforeActive afterActive
|
|
|
|
resetAuthority database 100
|
|
putResult <- runPut
|
|
env
|
|
user
|
|
Shared
|
|
(RelativePath "store-down.bin")
|
|
(OperationId "store-down")
|
|
9
|
|
identityA
|
|
(source ["desired-a"])
|
|
assert "store-down put rejected" (isLeft putResult)
|
|
rows <- Db.query database
|
|
"SELECT count(*) FROM mutations WHERE operation_id = ?"
|
|
(Only ("store-down" :: Text))
|
|
:: IO [Only Int]
|
|
assertEqual "store-down rejection creates no protocol record" [Only 0] rows
|
|
putStrLn "protocol action covered: LoseObjectStore"
|
|
putStrLn "store-down rejection preserves expired epoch boundary"
|
|
putStrLn "store-down rejection creates no protocol record"
|
|
|
|
verifyStoreRestored :: S3 -> IO ()
|
|
verifyStoreRestored store = do
|
|
available <- objectStoreAvailable store
|
|
assert "object store restored" available
|
|
forM_
|
|
["shared/delete-store-down.bin", "shared/expired-delete-store-down.bin"]
|
|
$ \key -> do
|
|
body <- expectRight =<< getObjectBytes store (ObjectKey key)
|
|
assertEqual "store loss preserves delete target" "prior-one" body
|
|
forM_
|
|
[ "shared/store-down.bin"
|
|
, "shared/expired-store-down.bin"
|
|
, "shared/active-store-down.bin"
|
|
]
|
|
$ \key -> do
|
|
absent <- expectRight =<< headObject store (ObjectKey key)
|
|
assertEqual "store loss leaves fenced object unchanged" Nothing absent
|
|
putStrLn "protocol action covered: RestoreObjectStore"
|
|
checkedInvariant "ObjectStoreFailureFencesObjects"
|
|
|
|
prepareStoreFence :: Db -> S3 -> IO ()
|
|
prepareStoreFence database store = do
|
|
resetAuthority database 100
|
|
seedObject database store "delete-store-down.bin" "prior-one"
|
|
seedObject database store "expired-delete-store-down.bin" "prior-one"
|
|
_ <- Db.execute database
|
|
"UPDATE pools SET used_bytes = 18 WHERE pool_id = ?"
|
|
(Only (unPoolId sharedPool))
|
|
putStrLn "store fence fixtures ready"
|
|
|
|
preparePostgresTransition :: Config -> Db -> IO ()
|
|
preparePostgresTransition cfg database = do
|
|
resetAuthority database 100
|
|
preparedRecord <- admitPut
|
|
cfg database "postgres-transition" "postgres-transition.bin" 9 identityA
|
|
assert "PostgreSQL transition fixture prepared" (isPrepared preparedRecord)
|
|
assertAuthority database 1 9 0
|
|
putStrLn "PostgreSQL transition fixture ready"
|
|
|
|
verifyPostgresUnavailable :: Config -> IO ()
|
|
verifyPostgresUnavailable cfg = do
|
|
attempt <-
|
|
(try $
|
|
withDb cfg $ \database ->
|
|
(Db.query_ database "SELECT 1" :: IO [Only Int]))
|
|
:: IO (Either SomeException [Only Int])
|
|
assert "PostgreSQL lane service is unavailable" (isLeft attempt)
|
|
coveredAction "LosePostgres"
|
|
|
|
verifyPostgresRestored :: Db -> IO ()
|
|
verifyPostgresRestored database = do
|
|
records <- withTransaction database $ \connection ->
|
|
lookupMutation connection protocolUser Put (OperationId "postgres-transition")
|
|
record <- maybe (fail "PostgreSQL transition record missing") pure records
|
|
assertEqual "PostgreSQL transition state preserved" Prepared (mrState record)
|
|
assertAuthority database 1 9 0
|
|
applyingAndSettled <- withTransaction database $ \connection -> do
|
|
applying <- markApplying connection record
|
|
settled <- markSettledFromIdle connection record
|
|
pure (applying && settled)
|
|
assert "restored PostgreSQL advances preserved mutation" applyingAndSettled
|
|
finalize database record ObjectAbsent MutationFailed
|
|
assertAuthority database 0 0 0
|
|
coveredAction "RestorePostgres"
|
|
checkedInvariant "PostgresFailureFencesDurableState"
|
|
|
|
dispatch :: Db -> MutationRecord -> IO ()
|
|
dispatch database record = do
|
|
ok <- withTransaction database $ \connection -> do
|
|
applying <- markApplying connection record
|
|
pending <- markExternalPending connection record 10 5
|
|
pure (applying && pending)
|
|
assert "PrepareEffect/DispatchEffect" ok
|
|
coveredAction "PrepareEffect"
|
|
coveredAction "DispatchEffect"
|
|
|
|
settle :: Db -> MutationRecord -> IO ()
|
|
settle database record = do
|
|
ok <- withTransaction database $ \connection ->
|
|
markExternalSettled connection record
|
|
assert "StoreSettlement" ok
|
|
coveredAction "StoreSettlement"
|
|
|
|
finalize
|
|
:: Db
|
|
-> MutationRecord
|
|
-> ObjectObservation
|
|
-> MutationResult
|
|
-> IO ()
|
|
finalize database record observation expected = do
|
|
actual <- withTransaction database $ \connection ->
|
|
finalizeFromObservation connection record observation
|
|
assertEqual "FinalizeSuccess/FinalizeFailure" (Just expected) actual
|
|
coveredAction "ControllerAdvance"
|
|
case expected of
|
|
MutationSucceeded -> coveredAction "FinalizeSuccess"
|
|
MutationFailed -> coveredAction "FinalizeFailure"
|
|
_ -> fail "terminal finalization expected success or failure"
|
|
|
|
admitPut
|
|
:: Config
|
|
-> Db
|
|
-> Text
|
|
-> Text
|
|
-> Integer
|
|
-> ContentSha256
|
|
-> IO AdmitOutcome
|
|
admitPut cfg database operation path size identity =
|
|
withTransaction database $ \connection ->
|
|
admitMutation
|
|
connection
|
|
protocolUser
|
|
Put
|
|
(OperationId operation)
|
|
sharedPool
|
|
(ObjectKey ("shared/" <> path))
|
|
(RelativePath path)
|
|
size
|
|
(Just identity)
|
|
(cfgMaxObjectBytes cfg)
|
|
(cfgIdempotencyWindowSec cfg)
|
|
|
|
admitDelete :: Config -> Db -> Text -> Text -> IO AdmitOutcome
|
|
admitDelete cfg database operation path =
|
|
withTransaction database $ \connection ->
|
|
admitMutation
|
|
connection
|
|
protocolUser
|
|
Delete
|
|
(OperationId operation)
|
|
sharedPool
|
|
(ObjectKey ("shared/" <> path))
|
|
(RelativePath path)
|
|
0
|
|
Nothing
|
|
(cfgMaxObjectBytes cfg)
|
|
(cfgIdempotencyWindowSec cfg)
|
|
|
|
admitPersonal :: Config -> Db -> UserId -> Text -> IO AdmitOutcome
|
|
admitPersonal cfg database user operation =
|
|
withTransaction database $ \connection ->
|
|
admitMutation
|
|
connection
|
|
user
|
|
Put
|
|
(OperationId operation)
|
|
(personalPool user)
|
|
(ObjectKey ("users/" <> unUserId user <> "/file.bin"))
|
|
(RelativePath "file.bin")
|
|
9
|
|
(Just identityA)
|
|
(cfgMaxObjectBytes cfg)
|
|
(cfgIdempotencyWindowSec cfg)
|
|
|
|
seedObject :: Db -> S3 -> Text -> BL.ByteString -> IO ()
|
|
seedObject database store path body = do
|
|
let identity = contentIdentity body
|
|
size = toInteger (BL.length body)
|
|
key = ObjectKey ("shared/" <> path)
|
|
expectRight =<< putObject store key size (Just identity) body
|
|
_ <- Db.execute database
|
|
"UPDATE pools SET used_bytes = ? WHERE pool_id = ?"
|
|
(size, unPoolId sharedPool)
|
|
_ <- Db.execute database
|
|
"INSERT INTO objects_meta \
|
|
\(object_key, pool_id, size_bytes, content_sha256) VALUES (?,?,?,?)"
|
|
(unObjectKey key, unPoolId sharedPool, size, unContentSha256 identity)
|
|
pure ()
|
|
|
|
resetScenario :: Db -> S3 -> Text -> IO ()
|
|
resetScenario database store path = do
|
|
resetAuthority database (64 * 1024 * 1024)
|
|
expectRight =<< deleteObject store (ObjectKey ("shared/" <> path))
|
|
|
|
resetAuthority :: Db -> Integer -> IO ()
|
|
resetAuthority database limitBytes = do
|
|
_ <- Db.execute_ database
|
|
"TRUNCATE key_locks, mutations, objects_meta, pools CASCADE"
|
|
configurePoolLimit database sharedPool limitBytes
|
|
|
|
assertAuthority :: Db -> Int -> Integer -> Integer -> IO ()
|
|
assertAuthority database locks reserved used = do
|
|
rows <- Db.query_ database
|
|
"SELECT \
|
|
\(SELECT count(*) FROM key_locks), \
|
|
\reserved_bytes, used_bytes \
|
|
\FROM pools WHERE pool_id = 'shared'"
|
|
:: IO [(Int, Integer, Integer)]
|
|
assertEqual "locks, reservations, and accounting"
|
|
[(locks, reserved, used)]
|
|
rows
|
|
|
|
mutationSnapshot
|
|
:: Db
|
|
-> Text
|
|
-> IO [(Text, Text, Text, Text, Text, Text, Integer, Maybe Text, Text, Text)]
|
|
mutationSnapshot database operation =
|
|
Db.query database
|
|
"SELECT user_id, kind, operation_id, pool_id, object_key, relative_path, \
|
|
\declared_size, content_sha256, state, external_state \
|
|
\FROM mutations WHERE operation_id = ?"
|
|
(Only operation)
|
|
|
|
bounded :: Config -> String -> IO () -> IO ()
|
|
bounded cfg label action = do
|
|
started <- getMonotonicTimeNSec
|
|
action
|
|
finished <- getMonotonicTimeNSec
|
|
let elapsedMicros = (finished - started) `div` 1000
|
|
boundMicros = fromInteger ((cfgObjectRequestTimeoutSec cfg + 1) * 1000000)
|
|
assert (label <> " exceeded configured settlement bound")
|
|
(elapsedMicros <= boundMicros)
|
|
|
|
stableForSettlementBound :: Config -> IO () -> IO ()
|
|
stableForSettlementBound cfg action = do
|
|
started <- getMonotonicTimeNSec
|
|
let duration =
|
|
fromInteger ((cfgObjectRequestTimeoutSec cfg + 1) * 1000000000)
|
|
observe = do
|
|
action
|
|
now <- getMonotonicTimeNSec
|
|
when (now - started < duration) $ do
|
|
threadDelay 100000
|
|
observe
|
|
observe
|
|
|
|
pacedSource :: MVar () -> BL.ByteString -> SourceT IO BS.ByteString
|
|
pacedSource started body =
|
|
SourceT $ \consume -> consume (go True (BL.toChunks body))
|
|
where
|
|
go _ [] = Stop
|
|
go first (chunk : chunks) = Effect $ do
|
|
when first $ do
|
|
putMVar started ()
|
|
threadDelay 300000
|
|
pure (Yield chunk (go False chunks))
|
|
|
|
spin :: IO a
|
|
spin = threadDelay 1000000 >> spin
|
|
|
|
prepared :: AdmitOutcome -> IO MutationRecord
|
|
prepared outcome =
|
|
case outcome of
|
|
AdmitPrepared record -> pure record
|
|
other -> fail ("expected prepared mutation, got " <> show other)
|
|
|
|
replayResult :: AdmitOutcome -> MutationResult
|
|
replayResult outcome =
|
|
case outcome of
|
|
AdmitReplay result _ -> result
|
|
other -> error ("expected replay, got " <> show other)
|
|
|
|
isPrepared :: AdmitOutcome -> Bool
|
|
isPrepared AdmitPrepared{} = True
|
|
isPrepared _ = False
|
|
|
|
isRejected :: AdmitOutcome -> Bool
|
|
isRejected AdmitRejected{} = True
|
|
isRejected _ = False
|
|
|
|
isConflict :: AdmitOutcome -> Bool
|
|
isConflict AdmitConflict = True
|
|
isConflict _ = False
|
|
|
|
isLeft :: Either left right -> Bool
|
|
isLeft Left{} = True
|
|
isLeft _ = False
|
|
|
|
contentIdentity :: BL.ByteString -> ContentSha256
|
|
contentIdentity =
|
|
ContentSha256
|
|
. TE.decodeUtf8
|
|
. B16.encode
|
|
. BA.convert
|
|
. (hashlazy :: BL.ByteString -> Digest SHA256)
|
|
|
|
expectRight :: Show left => Either left value -> IO value
|
|
expectRight result =
|
|
case result of
|
|
Left err -> fail ("unexpected failure: " <> show err)
|
|
Right value -> pure value
|
|
|
|
assert :: String -> Bool -> IO ()
|
|
assert label condition =
|
|
unless condition (fail label)
|
|
|
|
assertEqual :: (Eq value, Show value) => String -> value -> value -> IO ()
|
|
assertEqual label expected actual =
|
|
unless (expected == actual) $
|
|
fail (label <> ": expected " <> show expected <> ", got " <> show actual)
|
|
|
|
coveredAction :: Text -> IO ()
|
|
coveredAction action =
|
|
putStrLn ("protocol action covered: " <> T.unpack action)
|
|
|
|
checkedInvariant :: Text -> IO ()
|
|
checkedInvariant invariant =
|
|
putStrLn ("protocol invariant checked: " <> T.unpack invariant)
|
|
|
|
protocolUser :: UserId
|
|
protocolUser = UserId "protocol-user"
|
|
|
|
identityA :: ContentSha256
|
|
identityA =
|
|
ContentSha256
|
|
"bc14a1b33190bcb16e59f1c8a6f70dcd138e4e74fbdea4eca4f634780d5d6ccb"
|
|
|
|
identityB :: ContentSha256
|
|
identityB =
|
|
ContentSha256
|
|
"daeba8f55bf2b80cf87005e3687416f9cffbf23dcd4195bf13dbfbf6d5239a4e"
|