nuthatch/test/ProtocolConformance.hs

667 lines
25 KiB
Haskell
Raw Normal View History

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 "WP-10 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 "wp10" "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"