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.Encoding as TE import Database.PostgreSQL.Simple (Only (..)) import GHC.Clock (getMonotonicTimeNSec) import Nuthatch.App.Env import Nuthatch.Config import Nuthatch.Db import qualified Nuthatch.Db as Db import qualified Nuthatch.Mutation.Service as Mutation import Nuthatch.Mutation.Service (runDelete) import Nuthatch.ModelEvidence import Nuthatch.Protocol import Nuthatch.Redis import Nuthatch.S3 import Nuthatch.Transfer (inspectBandwidth, transferPieceSize) import Nuthatch.Types import Servant.Types.SourceT (SourceT (..), StepT (..), source) import System.Environment (getArgs) import System.IO (BufferMode (LineBuffering), hSetBuffering, stdout) runPut :: Env -> SessionUser -> Root -> RelativePath -> OperationId -> Integer -> ContentSha256 -> SourceT IO BS.ByteString -> IO (Either Mutation.PutFailure MutationResult) runPut env user root path operation size identity body = Mutation.runPut env user root path operation size identity (\key -> putObjectStream (envS3 env) key size (Just identity) (transferPieceSize env) (inspectBandwidth env user) body >>= pure . either (Left . Mutation.PutDispatchFailed) Right) main :: IO () main = do hSetBuffering stdout LineBuffering cfg <- loadConfig mode <- getArgs case mode of ["pg-fence"] -> verifyPostgresUnavailable cfg _ -> withDb (cfgPostgres cfg) $ \database -> do migrate database "sql" (cfgProduct cfg) withRedis (cfgRedis cfg) $ \redis -> do store <- newS3 (cfgS3 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 checkedModel CheckTypeOK 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 ActionFirstRequest coveredAction ActionReplayRequest coveredAction ActionConflictRequest checkedModel CheckRecordIdentityConsistent checkedModel CheckFingerprintBindsRequest checkedModel CheckObservationConsistent 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 checkedModel CheckReservationsConsistent checkedModel CheckLocksConsistent checkedModel CheckQuotaReservationBound 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 checkedModel CheckPersonalRequestsUseOwnPool checkedModel CheckPersonalPoolsDisjoint 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 ActionExternalEffectApplied checkedModel CheckActiveObjectConsistent checkedModel CheckExternalStateConsistent checkedModel CheckObjectStateCovered checkedModel CheckObjectQuotaBound checkedModel CheckAccountingAtQuiescence checkedModel CheckRecordsNeverForgotten checkedModel CheckRequestBindingsStable checkedModel CheckTerminalRecordsStable 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 ActionExternalEffectNotApplied 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 ActionLoseUploadBody coveredAction ActionSettleLostUpload checkedModel CheckUploadBodyConsistent checkedModel CheckUnfinishedEventuallyTerminal checkedModel CheckLostUploadEventuallyTerminal 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 checkedModel CheckSettledEffectsDoNotMutate 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 (cfgProduct 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" assertEqual "store-down expired delete epoch rejected" (Left (Mutation.DeleteDispatchFailed S3TransportError)) 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") assertEqual "store-down delete rejected" (Left (Mutation.DeleteDispatchFailed S3TransportError)) 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 (cfgProduct 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 coveredAction ActionLoseObjectStore 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 coveredAction ActionRestoreObjectStore checkedModel CheckObjectStoreFailureFencesObjects 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 (cfgPostgres cfg) $ \database -> (Db.query_ database "SELECT 1" :: IO [Only Int])) :: IO (Either SomeException [Only Int]) assert "PostgreSQL lane service is unavailable" (isLeft attempt) coveredAction ActionLosePostgres 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 ActionRestorePostgres checkedModel CheckPostgresFailureFencesDurableState dispatch :: Db -> MutationRecord -> IO () dispatch database record = do ok <- withTransaction database $ \connection -> do applying <- markApplying connection record pending <- markExternalPending connection record (DispatchTiming 10 5) pure (applying && pending) assert "PrepareEffect/DispatchEffect" ok coveredAction ActionPrepareEffect coveredAction ActionDispatchEffect settle :: Db -> MutationRecord -> IO () settle database record = do ok <- withTransaction database $ \connection -> markExternalSettled connection record assert "StoreSettlement" ok 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 case expected of MutationSucceeded -> coveredAction ActionFinalizeSuccess MutationFailed -> coveredAction ActionFinalizeFailure _ -> 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 AdmissionRequest { admissionIdentity = AdmissionIdentity protocolUser Put (OperationId operation) , admissionPoolId = sharedPool , admissionRelativePath = RelativePath path , admissionFingerprint = MutationFingerprint Put (ObjectKey ("shared/" <> path)) size (Just identity) } AdmissionLimits { admissionMaxObjectBytes = cfgMaxObjectBytes (cfgProduct cfg) , admissionIdempotencyWindowSec = cfgIdempotencyWindowSec (cfgProduct cfg) } admitDelete :: Config -> Db -> Text -> Text -> IO AdmitOutcome admitDelete cfg database operation path = withTransaction database $ \connection -> admitMutation connection AdmissionRequest { admissionIdentity = AdmissionIdentity protocolUser Delete (OperationId operation) , admissionPoolId = sharedPool , admissionRelativePath = RelativePath path , admissionFingerprint = MutationFingerprint Delete (ObjectKey ("shared/" <> path)) 0 Nothing } AdmissionLimits { admissionMaxObjectBytes = cfgMaxObjectBytes (cfgProduct cfg) , admissionIdempotencyWindowSec = cfgIdempotencyWindowSec (cfgProduct cfg) } admitPersonal :: Config -> Db -> UserId -> Text -> IO AdmitOutcome admitPersonal cfg database user operation = withTransaction database $ \connection -> admitMutation connection AdmissionRequest { admissionIdentity = AdmissionIdentity user Put (OperationId operation) , admissionPoolId = personalPool user , admissionRelativePath = RelativePath "file.bin" , admissionFingerprint = MutationFingerprint Put (ObjectKey ("users/" <> unUserId user <> "/file.bin")) 9 (Just identityA) } AdmissionLimits { admissionMaxObjectBytes = cfgMaxObjectBytes (cfgProduct cfg) , admissionIdempotencyWindowSec = cfgIdempotencyWindowSec (cfgProduct 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 (cfgRecovery 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 (cfgRecovery 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) protocolUser :: UserId protocolUser = UserId "protocol-user" identityA :: ContentSha256 identityA = ContentSha256 "bc14a1b33190bcb16e59f1c8a6f70dcd138e4e74fbdea4eca4f634780d5d6ccb" identityB :: ContentSha256 identityB = ContentSha256 "daeba8f55bf2b80cf87005e3687416f9cffbf23dcd4195bf13dbfbf6d5239a4e"