module Main (main) where
import Control.Concurrent (forkIO, killThread, threadDelay)
import Control.Concurrent.MVar (newEmptyMVar, putMVar, readMVar, takeMVar, tryReadMVar)
import Control.Exception (AsyncException (ThreadKilled), SomeException, bracket, bracket_, fromException, try)
import Control.Monad (forM, forM_)
import Data.ByteString.Char8 qualified as ByteString
import Data.IORef (IORef, modifyIORef', newIORef, readIORef)
import Data.List (find, nub, sort)
import Data.Maybe (catMaybes)
import Data.Text qualified as Text
import Data.Text.Encoding qualified as Encoding
import Database.PostgreSQL.LibPQ qualified as PQ
import GHC.Stack (HasCallStack, callStack, prettyCallStack)
import Hinagata.Config
import Hinagata.Connection
import Hinagata.Fixture.Bundle
import Hinagata.Fixture.Types
import Hinagata.Postgres.Baseline hiding (elapsedMs)
import Hinagata.Postgres.Cleanup
import Hinagata.Postgres.Error (FixtureLoadContext (..), LoadError (..), LoadPhase (..), loadErrorContext)
import Hinagata.Postgres.Lease
import Hinagata.Postgres.Load
import Hinagata.Postgres.Manager
import Hinagata.Postgres.Ownership
import Hinagata.Postgres.Session
import Hinagata.Prelude
import Hinagata.Types
import Paths_hinagata_postgres (getDataFileName)
import System.Directory (createDirectoryIfMissing, doesFileExist, renameFile)
import System.Environment (getArgs, getEnvironment, getExecutablePath, lookupEnv)
import System.Exit (ExitCode (..), exitFailure)
import System.FilePath ((</>))
import System.IO (IOMode (ReadWriteMode, WriteMode), withBinaryFile)
import System.IO.Temp (withSystemTempDirectory)
import System.Posix.Signals (sigKILL, signalProcess)
import System.Process (CreateProcess (..), ProcessHandle, getPid, getProcessExitCode, proc, waitForProcess, withCreateProcess)
import System.Timeout (timeout)
import TastyAdapter qualified
main :: IO ()
main = do
arguments <- getArgs
case arguments of
["--crash-lease-worker", fixtureRoot, bundleRoot, mode, readyFile] -> crashLeaseWorker fixtureRoot bundleRoot mode readyFile
["--acquisition-cancellation"] -> acquisitionCancellationTests
["--endpoint-worker", fixtureRoot, bundleRoot, selectedHost, expected] -> do
(target, _, _, _, _) <- environmentTarget selectedHost
plan <- bundleOrFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixtureName "endpoint-probe"]
result <- loadInto target defaultSessionOptions plan
assert "isolated explicit endpoint load" (case (expected, result) of ("success", Right _) -> True; ("failure", Left _) -> True; _ -> False)
["--lifecycle-worker", fixtureRoot, bundleRoot] -> do
socket <- requiredEnv "HINAGATA_TEST_PGHOST"
acquisitionCancellationTests
(target, host, port, user, database) <- environmentTarget socket
bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))) PQ.finish $ \connection -> do
plan <- bundleOrFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixtureName "good"]
lifecycleTests target connection host port user database fixtureRoot bundleRoot plan
[] -> do
host <- lookupEnv "HINAGATA_TEST_PGHOST"
case host of
Nothing -> putStrLn "PostgreSQL integration tests require scripts/test-postgres.sh"
Just socket -> integrationTests socket
_ -> fail "unknown integration-test arguments"
integrationTests :: FilePath -> IO ()
integrationTests socket = withSystemTempDirectory "hinagata-postgres-test-" $ \workspace -> do
portText <- requiredEnv "HINAGATA_TEST_PGPORT"
user <- requiredEnv "HINAGATA_TEST_PGUSER"
databaseText <- requiredEnv "HINAGATA_TEST_PGDATABASE"
ownedCluster <- (== Just "1") <$> lookupEnv "HINAGATA_TEST_OWNED_CLUSTER"
host <- orFail (socketDirectory (Text.pack socket))
port <- orFail (mkPort (read portText))
database <- orFail (mkDatabaseName (Text.pack databaseText))
target <- orFail (mkConnectionTarget host port (Text.pack user) database Nothing)
let fixtureRoot = workspace </> "fixtures"
bundleRoot = workspace </> "bundles"
config = BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)
goodDirectory = fixtureRoot </> "good"
badDirectory = fixtureRoot </> "bad"
mixedDirectory = fixtureRoot </> "mixed"
badCopyDirectory = fixtureRoot </> "bad-copy"
reuseDirectory = fixtureRoot </> "reuse"
deferredDirectory = fixtureRoot </> "deferred"
slowDirectory = fixtureRoot </> "slow"
slowCopyDirectory = fixtureRoot </> "slow-copy"
sequenceDirectory = fixtureRoot </> "sequence"
analyzeDirectory = fixtureRoot </> "analyze"
finalCopyDirectory = fixtureRoot </> "final-copy-error"
createDirectoryIfMissing True goodDirectory
createDirectoryIfMissing True badDirectory
createDirectoryIfMissing True mixedDirectory
createDirectoryIfMissing True badCopyDirectory
createDirectoryIfMissing True reuseDirectory
createDirectoryIfMissing True deferredDirectory
createDirectoryIfMissing True slowDirectory
createDirectoryIfMissing True slowCopyDirectory
createDirectoryIfMissing True sequenceDirectory
createDirectoryIfMissing True analyzeDirectory
createDirectoryIfMissing True finalCopyDirectory
ByteString.writeFile (goodDirectory </> "fixture.sql") "INSERT INTO items (id, note) VALUES (1, 'first'); INSERT INTO items (id, note) VALUES (2, 'second');"
ByteString.writeFile (badDirectory </> "fixture.yaml") "name: bad\nsteps:\n - sql: first.sql\n - sql: fail.sql\n"
ByteString.writeFile (badDirectory </> "first.sql") "INSERT INTO items (id, note) VALUES (3, 'rolled back');"
ByteString.writeFile (badDirectory </> "fail.sql") "INSERT INTO items (id, note) VALUES (1, 'duplicate');"
ByteString.writeFile (mixedDirectory </> "fixture.yaml") "name: mixed\nsteps:\n - sql: before.sql\n - copy:\n table: {schema: public, name: items}\n columns: [id, note]\n file: rows.csv\n format: csv\n header: false\n - sql: after.sql\n"
ByteString.writeFile (mixedDirectory </> "before.sql") "INSERT INTO items (id, note) VALUES (4, 'before');"
ByteString.writeFile (mixedDirectory </> "rows.csv") "5,from csv\n6,from csv\n"
ByteString.writeFile (mixedDirectory </> "after.sql") "UPDATE items SET note = 'after' WHERE id = 6;"
ByteString.writeFile (badCopyDirectory </> "fixture.yaml") "name: bad-copy\nsteps:\n - sql: before.sql\n - copy:\n table: {schema: public, name: items}\n columns: [id, note]\n file: rows.csv\n format: csv\n header: false\n"
ByteString.writeFile (badCopyDirectory </> "before.sql") "INSERT INTO items (id, note) VALUES (7, 'rolled back');"
ByteString.writeFile (badCopyDirectory </> "rows.csv") "8,first row\n1,duplicate key\n"
ByteString.writeFile (reuseDirectory </> "fixture.sql") "INSERT INTO items (id, note) VALUES (10, 'reused session');"
ByteString.writeFile (deferredDirectory </> "fixture.sql") "INSERT INTO child_items (parent_id) VALUES (404);"
ByteString.writeFile (slowDirectory </> "fixture.sql") "SELECT pg_sleep(1);"
ByteString.writeFile (slowCopyDirectory </> "fixture.yaml") "name: slow-copy\nsteps:\n - copy:\n table: {schema: public, name: slow_copy_items}\n columns: [id]\n file: rows.csv\n format: csv\n header: false\n"
ByteString.writeFile (slowCopyDirectory </> "rows.csv") (ByteString.pack (unlines (map show [1 .. 200 :: Int])))
ByteString.writeFile (sequenceDirectory </> "fixture.sql") "INSERT INTO sequence_items (id, note) VALUES (500, 'explicit'); SELECT setval(pg_get_serial_sequence('public.sequence_items', 'id'), 500, true);"
ByteString.writeFile (analyzeDirectory </> "fixture.yaml") "name: analyze\nsteps:\n - copy:\n table: {schema: public, name: items}\n columns: [id, note]\n file: rows.csv\n format: csv\n header: false\n - sql: analyze.sql\n"
ByteString.writeFile (analyzeDirectory </> "rows.csv") "11,bulk row\n12,bulk row\n"
ByteString.writeFile (analyzeDirectory </> "analyze.sql") "ANALYZE items;"
ByteString.writeFile (finalCopyDirectory </> "fixture.yaml") "name: final-copy-error\nsteps:\n - copy:\n table: {schema: public, name: final_copy_items}\n columns: [id]\n file: rows.csv\n format: csv\n header: false\n"
ByteString.writeFile (finalCopyDirectory </> "rows.csv") "1\n2\n"
when ownedCluster (runEndpointWorker workspace "127.0.0.2" ["--lifecycle-worker", fixtureRoot, bundleRoot])
setupConnection <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))
execCheck setupConnection "CREATE TABLE items (id integer PRIMARY KEY, note text NOT NULL)"
execCheck setupConnection "INSERT INTO items (id, note) VALUES (99, 'sentinel')"
execCheck setupConnection "CREATE TABLE parent_items (id integer PRIMARY KEY)"
execCheck setupConnection "CREATE TABLE child_items (parent_id integer REFERENCES parent_items(id) DEFERRABLE INITIALLY DEFERRED)"
execCheck setupConnection "CREATE TABLE slow_copy_items (id integer PRIMARY KEY)"
execCheck setupConnection "CREATE FUNCTION slow_copy_row() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(0.01); RETURN NEW; END $$"
execCheck setupConnection "CREATE TRIGGER slow_copy_row BEFORE INSERT ON slow_copy_items FOR EACH ROW EXECUTE FUNCTION slow_copy_row()"
execCheck setupConnection "CREATE TABLE sequence_items (id bigserial PRIMARY KEY, note text NOT NULL)"
execCheck setupConnection "CREATE TABLE final_copy_items (id integer PRIMARY KEY)"
execCheck setupConnection "CREATE FUNCTION fail_final_copy() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'final COPY failure'; END $$"
execCheck setupConnection "CREATE TRIGGER fail_final_copy AFTER INSERT ON final_copy_items FOR EACH STATEMENT EXECUTE FUNCTION fail_final_copy()"
let bypassDirectory = fixtureRoot </> "whitespace-bypass"
createDirectoryIfMissing True bypassDirectory
ByteString.writeFile (bypassDirectory </> "fixture.yaml") "name: whitespace-bypass\nsteps:\n - sql: first.sql\n - sql: commit.sql\n - sql: fail.sql\n"
ByteString.writeFile (bypassDirectory </> "first.sql") "CREATE TABLE escaped_transaction (id integer); INSERT INTO escaped_transaction VALUES (1); UPDATE items SET note = 'changed' WHERE id = 99;"
ByteString.writeFile (bypassDirectory </> "commit.sql") "\fCOMMIT;"
ByteString.writeFile (bypassDirectory </> "fail.sql") "SELECT 1/0;"
bypass <- compileFixtures config [fixtureName "whitespace-bypass"]
-- Follow the public compile/load workflow even on the vulnerable implementation.
case bypass of
Right plan -> void (loadInto target defaultSessionOptions plan)
Left _ -> pure ()
assert "form-feed COMMIT rejected during capture" (case bypass of Left (InvalidSql _ _) -> True; _ -> False)
assert "rejected closure preserves caller sentinel" =<< ((== 1) <$> queryInt setupConnection "SELECT count(*) FROM items WHERE id = 99 AND note = 'sentinel'")
assert "rejected closure creates no table" =<< ((== 0) <$> queryInt setupConnection "SELECT count(*) FROM pg_class WHERE relname = 'escaped_transaction'")
goodPlan <- bundleOrFail =<< compileFixtures config [fixtureName "good"]
endpointTests workspace socket setupConnection fixtureRoot bundleRoot
loaded <- loadInto target defaultSessionOptions goodPlan
assert "SQL closure loads" (case loaded of Right report -> fixtureCount report == 1 && stepCount report == 1; Left _ -> False)
assert "SQL rows committed" =<< ((== 3) <$> queryInt setupConnection "SELECT count(*) FROM items")
badPlan <- bundleOrFail =<< compileFixtures config [fixtureName "bad"]
failed <- loadInto target defaultSessionOptions badPlan
assert "late SQL error reported with SQLSTATE" (case failed of Left LoadError {sqlState = Just "23505", targetIdentity = Just _} -> True; _ -> False)
assert "failed closure rolled back" =<< ((== 3) <$> queryInt setupConnection "SELECT count(*) FROM items")
assert "sentinel remains" =<< ((== 1) <$> queryInt setupConnection "SELECT count(*) FROM items WHERE id = 99 AND note = 'sentinel'")
mixedPlan <- bundleOrFail =<< compileFixtures config [fixtureName "mixed"]
mixedLoaded <- loadInto target defaultSessionOptions mixedPlan
assert "mixed SQL/COPY/SQL closure loads" (case mixedLoaded of Right report -> stepCount report == 3 && bytesSent report > 0; Left _ -> False)
assert "mixed closure committed" =<< ((== 6) <$> queryInt setupConnection "SELECT count(*) FROM items")
assert "post-COPY SQL ordered" =<< ((== 1) <$> queryInt setupConnection "SELECT count(*) FROM items WHERE id = 6 AND note = 'after'")
badCopyPlan <- bundleOrFail =<< compileFixtures config [fixtureName "bad-copy"]
badCopyResult <- loadInto target defaultSessionOptions badCopyPlan
assert "late COPY error reported" (case badCopyResult of Left _ -> True; Right _ -> False)
assert "failed COPY rolls back prior SQL and rows" =<< ((== 6) <$> queryInt setupConnection "SELECT count(*) FROM items")
reusePlan <- bundleOrFail =<< compileFixtures config [fixtureName "reuse"]
reused <- withSession target defaultSessionOptions $ \session -> do
first <- loadPlan session badPlan
second <- loadPlan session reusePlan
pure (first, second)
assert "session reuses after rollback" (case reused of Right (Left _, Right _) -> True; _ -> False)
assert "reused session committed" =<< ((== 7) <$> queryInt setupConnection "SELECT count(*) FROM items")
deferredPlan <- bundleOrFail =<< compileFixtures config [fixtureName "deferred"]
deferred <- loadInto target defaultSessionOptions deferredPlan
assert "deferred constraint failure reported at COMMIT" (case deferred of Left LoadError {phase = CommitLoad} -> True; _ -> False)
assert "deferred failure did not insert" =<< ((== 0) <$> queryInt setupConnection "SELECT count(*) FROM child_items")
ByteString.writeFile (planDirectory mixedPlan </> "step-1") "corrupted"
corrupt <- loadInto target defaultSessionOptions mixedPlan
assert "modified bundle refused before mutation" (case corrupt of Left LoadError {phase = VerifyBundle} -> True; _ -> False)
assert "modified bundle left target intact" =<< ((== 7) <$> queryInt setupConnection "SELECT count(*) FROM items")
slowPlan <- bundleOrFail =<< compileFixtures config [fixtureName "slow"]
concurrent <- withSession target defaultSessionOptions $ \session -> do
finished <- newEmptyMVar
_ <- forkIO (loadPlan session slowPlan >>= putMVar finished)
threadDelay 100000
simultaneous <- loadPlan session reusePlan
first <- timeout 3000000 (takeMVar finished)
pure (simultaneous, first)
assert "simultaneous session use refused" (case concurrent of Right (Left LoadError {cause = "SessionBusy"}, Just (Right _)) -> True; _ -> False)
escaped <- withSession target defaultSessionOptions pure
afterClose <- case escaped of
Left failure -> fail (show failure)
Right closed -> loadPlan closed slowPlan
assert "closed session refused" (case afterClose of Left LoadError {cause = "SessionClosed"} -> True; _ -> False)
tcp <- orFail (tcpHost "127.0.0.1")
tcpTarget <- orFail (mkConnectionTarget tcp port (Text.pack user) database Nothing)
tcpEnabled <- lookupEnv "HINAGATA_TEST_TCP"
case tcpEnabled of
Just "1" -> do
tcpResult <- loadInto tcpTarget defaultSessionOptions slowPlan
assert "TCP target loads without socket fallback" (case tcpResult of Right _ -> True; Left _ -> False)
_ -> do
tcpAttempt <- withSession tcpTarget defaultSessionOptions (const (pure ()))
assert "socket-only cluster does not silently retry TCP" (case tcpAttempt of Left _ -> True; Right _ -> False)
missingHost <- orFail (socketDirectory (Text.pack (socket </> "missing")))
missingTarget <- orFail (mkConnectionTarget missingHost port (Text.pack user) database Nothing)
missingSocket <- withSession missingTarget defaultSessionOptions (const (pure ()))
assert "missing socket fails without a TCP fallback" (case missingSocket of Left _ -> True; Right _ -> False)
slowCopyPlan <- bundleOrFail =<< compileFixtures config [fixtureName "slow-copy"]
cancelled <- withSession target defaultSessionOptions $ \session -> do
completed <- newEmptyMVar
worker <- forkIO (try @SomeException (loadPlan session slowCopyPlan) >>= putMVar completed)
threadDelay 100000
killThread worker
interrupted <- timeout 3000000 (takeMVar completed)
afterInterrupt <- loadPlan session slowCopyPlan
pure (interrupted, afterInterrupt)
assert "COPY interruption closes the session" (case cancelled of Right (Just (Left _), Left LoadError {cause = "SessionClosed"}) -> True; _ -> False)
assert "cancelled COPY left no rows" =<< ((== 0) <$> queryInt setupConnection "SELECT count(*) FROM slow_copy_items")
detached <- waitForNoLoader setupConnection 5000000
assert "cancelled loader connection exits within cleanup deadline" detached
disconnected <- withSession target defaultSessionOptions $ \session -> do
completed <- newEmptyMVar
_ <- forkIO (try @SomeException (loadPlan session slowPlan) >>= putMVar completed)
threadDelay 100000
terminated <- PQ.exec setupConnection "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = current_database() AND pid <> pg_backend_pid()"
assert "server disconnect request succeeded" (case terminated of Just _ -> True; Nothing -> False)
result <- timeout 3000000 (takeMVar completed)
retry <- loadPlan session slowPlan
pure (result, retry)
assert "server disconnect retires session" (case disconnected of Right (Just (Right (Left _)), Left LoadError {cause = "SessionClosed"}) -> True; Right (Just (Left _), Left LoadError {cause = "SessionClosed"}) -> True; _ -> False)
sequencePlan <- bundleOrFail =<< compileFixtures config [fixtureName "sequence"]
sequenced <- loadInto target defaultSessionOptions sequencePlan
assert "explicit ID and sequence adjustment load" (case sequenced of Right _ -> True; _ -> False)
generatedId <- queryInt setupConnection "INSERT INTO sequence_items (note) VALUES ('generated') RETURNING id"
assert "application insert follows explicit ID" (generatedId == 501)
analyzePlan <- bundleOrFail =<< compileFixtures config [fixtureName "analyze"]
analyzed <- loadInto target defaultSessionOptions analyzePlan
assert "explicit ANALYZE follows COPY and has separate stage timings" (case analyzed of Right report -> stepCount report == 2 && bytesSent report > 0 && copyMs report <= elapsedMs report && sqlMs report <= elapsedMs report; _ -> False)
assert "ANALYZE fixture committed" =<< ((== 9) <$> queryInt setupConnection "SELECT count(*) FROM items")
finalCopyPlan <- bundleOrFail =<< compileFixtures config [fixtureName "final-copy-error"]
finalCopy <- loadInto target defaultSessionOptions finalCopyPlan
assert "final COPY result error is reported" (case finalCopy of Left LoadError {phase = ExecuteCopy, sqlState = Just "P0001"} -> True; _ -> False)
assert "final COPY result error rolls back" =<< ((== 0) <$> queryInt setupConnection "SELECT count(*) FROM final_copy_items")
PQ.finish setupConnection
acquisitionCancellationTests :: IO ()
acquisitionCancellationTests = forM_ [False, True] $ \useTimeout ->
withSystemTempDirectory "hinagata-stalled-peer-" $ \directory -> do
peerScript <- getDataFileName "test/stalled-postgres-peer.py"
withCreateProcess (proc "python3" [peerScript, directory]) $ \_ _ _ peer -> do
waitForChild peer (doesFileExist (directory </> "port") >>= \ready -> pure (if ready then Just () else Nothing))
port <- orFail . mkPort . read =<< readFile (directory </> "port")
host <- orFail (tcpHost "127.0.0.1")
database <- orFail (mkDatabaseName "unused")
target <- orFail (mkConnectionTarget host port "unused" database Nothing)
finished <- newEmptyMVar
worker <-
forkIO $
try @SomeException
( if useTimeout
then void (timeout 500000 (withSession target defaultSessionOptions (const (pure ()))))
else void (withSession target defaultSessionOptions (const (pure ())))
)
>>= putMVar finished
waitForChild peer (doesFileExist (directory </> "accepted") >>= \ready -> pure (if ready then Just () else Nothing))
unless useTimeout (killThread worker)
result <- timeout 2000000 (takeMVar finished)
assert "stalled connection cancellation returns promptly" (case (useTimeout, result) of (True, Just (Right ())) -> True; (False, Just (Left exception)) -> fromException exception == Just ThreadKilled; _ -> False)
closed <- timeout 1000000 (waitForProcess peer)
assert "cancelled acquisition closes native socket without GC" (closed == Just ExitSuccess)
assert "stalled peer records EOF" =<< doesFileExist (directory </> "closed")
-- Child environments keep conflicting libpq defaults away from concurrent tests.
environmentTarget :: FilePath -> IO (ConnectionTarget, Host, Port, Text, DatabaseName)
environmentTarget selectedHost = do
host <- orFail (if take 1 selectedHost == "/" then socketDirectory (Text.pack selectedHost) else tcpHost (Text.pack selectedHost))
port <- orFail . mkPort . read =<< requiredEnv "HINAGATA_TEST_PGPORT"
user <- Text.pack <$> requiredEnv "HINAGATA_TEST_PGUSER"
database <- orFail . mkDatabaseName . Text.pack =<< requiredEnv "HINAGATA_TEST_PGDATABASE"
target <- orFail (mkConnectionTarget host port user database Nothing)
pure (target, host, port, user, database)
runEndpointWorker :: FilePath -> String -> [String] -> IO ()
runEndpointWorker workspace address arguments = do
executable <- getExecutablePath
inherited <- getEnvironment
let serviceFile = workspace </> "pg_service.conf"
overlay = [("PGHOSTADDR", address), ("PGHOST", "/missing/ambient/socket"), ("PGPORT", "1"), ("PGUSER", "missing_ambient_role"), ("PGDATABASE", "missing_ambient_database"), ("PGSERVICE", "conflict"), ("PGSERVICEFILE", serviceFile)]
command = (proc executable arguments) {env = Just (overlay ++ filter (\(key, _) -> key `notElem` map fst overlay) inherited)}
ByteString.writeFile serviceFile "[conflict]\nhostaddr=127.0.0.2\nhost=/missing/service/socket\nport=1\nuser=missing_service_role\ndbname=missing_service_database\n"
withCreateProcess command $ \_ _ _ child -> do
status <- timeout 180000000 (waitForProcess child)
assert "endpoint subprocess succeeds within bound" (status == Just ExitSuccess)
after <- getEnvironment
assert "endpoint subprocesses preserve parent environment" (sort inherited == sort after)
endpointTests :: FilePath -> FilePath -> PQ.Connection -> FilePath -> FilePath -> IO ()
endpointTests workspace socket connection fixtureRoot bundleRoot = do
let directory = fixtureRoot </> "endpoint-probe"
run = runEndpointWorker workspace
createDirectoryIfMissing True directory
ByteString.writeFile (directory </> "fixture.sql") "INSERT INTO endpoint_probe VALUES (1);"
execCheck connection "CREATE TABLE endpoint_probe (id integer)"
start <- newEmptyMVar
completed <- forM [("valid", "127.0.0.2", socket, "success"), ("missing", "127.0.0.1", socket </> "missing", "failure")] $ \(label, address, selectedHost, expected) -> do
let private = workspace </> ("endpoint-" ++ label)
createDirectoryIfMissing True private
finished <- newEmptyMVar
_ <-
forkIO $
try @SomeException
(readMVar start >> runEndpointWorker private address ["--endpoint-worker", fixtureRoot, private </> "bundles", selectedHost, expected])
>>= putMVar finished
pure finished
putMVar start ()
outcomes <- mapM takeMVar completed
assert "concurrent explicit endpoint workers stay independent" (all (either (const False) (const True)) outcomes)
run "" ["--endpoint-worker", fixtureRoot, bundleRoot, socket, "success"]
assert "missing socket never dispatches to reachable TCP" =<< ((== 2) <$> queryInt connection "SELECT count(*) FROM endpoint_probe")
tcpEnabled <- lookupEnv "HINAGATA_TEST_TCP"
when (tcpEnabled == Just "1") $ do
run "127.0.0.2" ["--endpoint-worker", fixtureRoot, bundleRoot, "127.0.0.1", "success"]
assert "explicit TCP overrides ambient address" =<< ((== 3) <$> queryInt connection "SELECT count(*) FROM endpoint_probe")
lifecycleConfigFor :: Host -> Port -> Text -> DatabaseName -> FilePath -> FilePath -> SqlIdentifier -> ProjectId -> SqlIdentifier -> SqlIdentifier -> HinagataConfig
lifecycleConfigFor host port adminUser database fixtureRoot bundleRoot catalogSchema projectId publicSchema messageSetting =
HinagataConfig
{ endpoint = EndpointConfig host port,
project = projectId,
fixtureRoot,
bundleRoot,
maintenanceDatabase = database,
maintenanceSchema = catalogSchema,
administration = AccessConfig adminUser database Nothing,
setup = AccessConfig "hinagata_setup" database Nothing,
application = AccessConfig "hinagata_app" database Nothing,
setupGrants = [GrantConnect, GrantCreate, GrantTemporary],
applicationGrants = [GrantConnect],
setupSchemaGrants = [SchemaGrant publicSchema SchemaUsage, SchemaGrant publicSchema SchemaCreate],
applicationSchemaGrants = [SchemaGrant publicSchema SchemaUsage],
setupSettings = [RoleSetting messageSetting "warning"],
applicationSettings = [RoleSetting messageSetting "warning"],
cloneStrategy = WalLog,
acquisitionDeadlineMs = positive 300000,
setupDeadlineMs = positive 300000,
setupWorkers = positive 2,
activeLeases = positive 4,
pendingRequests = positive 8,
chunkSize = positive 65536,
sqlSizeLimit = positive 1048576
}
crashLeaseWorker :: FilePath -> FilePath -> String -> FilePath -> IO ()
crashLeaseWorker fixtureRoot bundleRoot mode readyFile = do
socket <- requiredEnv "HINAGATA_TEST_PGHOST"
portText <- requiredEnv "HINAGATA_TEST_PGPORT"
adminUser <- Text.pack <$> requiredEnv "HINAGATA_TEST_PGUSER"
database <- orFail . mkDatabaseName . Text.pack =<< requiredEnv "HINAGATA_TEST_PGDATABASE"
host <- orFail (socketDirectory (Text.pack socket))
port <- orFail (mkPort (read portText))
catalogSchema <- orFail (mkSqlIdentifier "hinagata_test")
projectId <- orFail (mkProjectId "baseline-test")
publicSchema <- orFail (mkSqlIdentifier "public")
messageSetting <- orFail (mkSqlIdentifier "client_min_messages")
let configuration = lifecycleConfigFor host port adminUser database fixtureRoot bundleRoot catalogSchema projectId publicSchema messageSetting
bundleConfig = BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)
base <- bundleOrFail =<< compileFixtures bundleConfig [fixtureName "good"]
let unexpectedHook _ = pure (Left "crash worker unexpectedly rebuilt the baseline")
specification = BaselineSpec configuration base (Just "migration-v1") (Just "verify-v1") ["plpgsql"] (Just "C") unexpectedHook unexpectedHook Nothing Nothing
case mode of
"building" -> do
let blockedHook _ = writeReady readyFile "building" >> threadDelay 60000000 >> pure (Right ())
_ <- ensureBaseline specification {migrationRevision = Just "migration-process-crash", migrationHook = blockedHook}
fail "crash worker reached normal baseline completion"
"sealing" -> buildAtPublication specification "migration-seal-crash" False
"published" -> buildAtPublication specification "migration-published-crash" True
"intent" -> runLease specification bundleConfig
"created" -> runLease specification bundleConfig
"binding" -> runLease specification bundleConfig
"drop" -> runLease specification bundleConfig
"handoff" -> runLease specification bundleConfig
"loading" -> runLease specification bundleConfig
_ -> fail "unknown crash-worker phase"
where
buildAtPublication specification revision reportReady = do
let migrate target = do
migrated <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))
execCheck migrated "CREATE TABLE items (id integer PRIMARY KEY, note text NOT NULL)"
execCheck migrated "GRANT SELECT, INSERT ON items TO hinagata_app"
PQ.finish migrated
pure (Right ())
verify target = do
verified <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))
count <- queryInt verified "SELECT count(*) FROM items"
PQ.finish verified
pure (if count == 2 then Right () else Left "unexpected baseline rows")
(baseline, report) <- orFail =<< ensureBaseline specification {migrationRevision = Just revision, migrationHook = migrate, verificationHook = verify}
assert "publication worker built a new baseline" (kind report == Built)
when reportReady (writeReady readyFile (Encoding.encodeUtf8 (databaseNameText (baselineDatabase baseline))))
threadDelay 60000000
runLease specification bundleConfig = do
(baseline, report) <- orFail =<< ensureBaseline specification
assert "crash worker reuses the existing sealed baseline" (kind report == Reused)
let scenarioName = if mode == "loading" then "crash-loading" else "scenario"
scenario <- bundleOrFail =<< compileFixtures bundleConfig [fixtureName "good", fixtureName scenarioName]
_ <- withDatabase (configuration specification) baseline scenario $ \LeaseInfo {leaseId} -> do
when (mode == "handoff") (writeReady readyFile (Encoding.encodeUtf8 leaseId))
if mode == "drop"
then writeReady readyFile (Encoding.encodeUtf8 leaseId) >> waitForGo (readyFile ++ ".go")
else threadDelay 60000000
fail "crash worker reached normal lease completion"
waitForGo path = do
present <- doesFileExist path
unless present (threadDelay 10000 >> waitForGo path)
writeReady :: FilePath -> ByteString.ByteString -> IO ()
writeReady path payload = do
let temporary = path ++ ".tmp"
ByteString.writeFile temporary payload
renameFile temporary path
lifecycleTests :: ConnectionTarget -> PQ.Connection -> Host -> Port -> Text -> DatabaseName -> FilePath -> FilePath -> FixturePlan -> IO ()
lifecycleTests target connection host port adminUser database fixtureRoot bundleRoot goodPlan = do
legacySchema <- orFail (mkSqlIdentifier "hinagata_legacy_test")
legacyDdl <- ByteString.readFile =<< getDataFileName "sql/catalog-v1.sql"
execCheck connection (Encoding.encodeUtf8 (Text.replace "%SCHEMA%" (quoteSqlIdentifier legacySchema) (Encoding.decodeUtf8 legacyDdl)))
legacyVersion <- queryInt connection "SELECT format_version FROM hinagata_legacy_test.meta"
assert "legacy fixture starts at catalog version 1" (legacyVersion == 1)
legacyPreview <- inspectCatalog target legacySchema defaultSessionOptions
assert "read-only inspection requests a legacy catalog upgrade" (case legacyPreview of Left CatalogError {cause} -> "version-2 upgrade" `Text.isInfixOf` cause; _ -> False)
legacyUuid <- queryText connection "SELECT cluster_uuid::text FROM hinagata_legacy_test.meta"
upgraded <- ensureCatalog target legacySchema defaultSessionOptions
assert "owned version-1 catalog upgrades in place" (case upgraded of Right CatalogIdentity {formatVersion = 2, clusterUuid} -> Encoding.encodeUtf8 clusterUuid == legacyUuid; _ -> False)
inspectedUpgrade <- inspectCatalog target legacySchema defaultSessionOptions
assert "upgraded catalog passes read-only inspection" (inspectedUpgrade == upgraded)
assert "upgrade installs diagnostic columns" =<< ((== 2) <$> queryInt connection "SELECT count(*) FROM information_schema.columns WHERE table_schema = 'hinagata_legacy_test' AND table_name IN ('generations', 'allocations') AND column_name = 'last_error'")
execCheck connection "ALTER TABLE hinagata_legacy_test.meta DROP CONSTRAINT meta_format_version_check"
execCheck connection "UPDATE hinagata_legacy_test.meta SET format_version = 99"
unknownCatalog <- ensureCatalog target legacySchema defaultSessionOptions
unknownPreview <- inspectCatalog target legacySchema defaultSessionOptions
assert "unknown catalog versions are refused without replacement" (case (unknownCatalog, unknownPreview) of (Left _, Left _) -> True; _ -> False)
assert "refused unknown catalog retains its version" =<< ((== 99) <$> queryInt connection "SELECT format_version FROM hinagata_legacy_test.meta")
execCheck connection "UPDATE hinagata_legacy_test.meta SET format_version = 2"
execCheck connection "ALTER TABLE hinagata_legacy_test.meta ADD CONSTRAINT meta_format_version_check CHECK (format_version = 2)"
partialSchema <- orFail (mkSqlIdentifier "hinagata_partial_test")
execCheck connection (Encoding.encodeUtf8 (Text.replace "%SCHEMA%" (quoteSqlIdentifier partialSchema) (Encoding.decodeUtf8 legacyDdl)))
execCheck connection "ALTER TABLE hinagata_partial_test.allocations ADD COLUMN last_error text"
refusedUpgrade <- ensureCatalog target partialSchema defaultSessionOptions
assert "failed catalog upgrade is refused" (case refusedUpgrade of Left _ -> True; Right _ -> False)
assert "failed upgrade rolls back format version" =<< ((== 1) <$> queryInt connection "SELECT format_version FROM hinagata_partial_test.meta")
assert "failed upgrade rolls back its earlier DDL" =<< ((== 0) <$> queryInt connection "SELECT count(*) FROM information_schema.columns WHERE table_schema = 'hinagata_partial_test' AND table_name = 'generations' AND column_name = 'last_error'")
catalogSchema <- orFail (mkSqlIdentifier "hinagata_test")
absentPreview <- inspectCatalog target catalogSchema defaultSessionOptions
assert "catalog inspection refuses an absent schema without creating it" (case absentPreview of Left _ -> True; Right _ -> False)
assert "catalog inspection leaves the schema absent" =<< ((== 0) <$> queryInt connection "SELECT count(*) FROM pg_namespace WHERE nspname = 'hinagata_test'")
firstCatalog <- ensureCatalog target catalogSchema defaultSessionOptions
secondCatalog <- ensureCatalog target catalogSchema defaultSessionOptions
assert "catalog initializes and reuses one cluster identity" (case (firstCatalog, secondCatalog) of (Right first, Right second) -> first == second && formatVersion first == 2; _ -> False)
catalog <- orFail firstCatalog
execCheck connection "CREATE DATABASE hinagata_owned_test"
let marker = "hinagata:v1:" <> clusterUuid catalog <> ":test-token"
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE hinagata_owned_test IS '" <> marker <> "'"))
ownedName <- orFail (mkDatabaseName "hinagata_owned_test")
ownedOid <- queryInt connection "SELECT oid FROM pg_database WHERE datname = 'hinagata_owned_test'"
let owned = OwnedDatabase {name = ownedName, oid = ownedOid, token = "test-token"}
matched <- verifyOwnedDatabase target defaultSessionOptions catalog owned
assert "name, OID, and marker prove ownership" (matched == Right OwnershipMatches)
wrongOid <- verifyOwnedDatabase target defaultSessionOptions catalog owned {oid = ownedOid + 1}
assert "matching name and marker cannot override a changed OID" (wrongOid == Right OwnershipMismatch)
wrongCluster <- verifyOwnedDatabase target defaultSessionOptions catalog {clusterUuid = "replaced-cluster"} owned
assert "records from a replaced cluster cannot authorize the database" (wrongCluster == Right OwnershipMismatch)
execCheck connection "COMMENT ON DATABASE hinagata_owned_test IS 'foreign'"
replaced <- verifyOwnedDatabase target defaultSessionOptions catalog owned
assert "changed ownership marker is refused" (replaced == Right OwnershipMismatch)
execCheck connection "DROP DATABASE hinagata_owned_test"
missing <- verifyOwnedDatabase target defaultSessionOptions catalog owned
assert "missing owned database is distinct from a mismatch" (missing == Right OwnershipMissing)
execCheck connection "CREATE SCHEMA foreign_catalog"
foreignSchema <- orFail (mkSqlIdentifier "foreign_catalog")
foreignCatalog <- ensureCatalog target foreignSchema defaultSessionOptions
assert "existing foreign schema without format marker is refused" (case foreignCatalog of Left _ -> True; _ -> False)
execCheck connection "CREATE ROLE hinagata_setup LOGIN"
execCheck connection "CREATE ROLE hinagata_app LOGIN"
projectId <- orFail (mkProjectId "baseline-test")
publicSchema <- orFail (mkSqlIdentifier "public")
messageSetting <- orFail (mkSqlIdentifier "client_min_messages")
migrationCalls <- newIORef (0 :: Int)
let lifecycleConfig = lifecycleConfigFor host port adminUser database fixtureRoot bundleRoot catalogSchema projectId publicSchema messageSetting
adminRole = AccessConfig adminUser database Nothing
setupRole = AccessConfig "hinagata_setup" database Nothing
appRole = AccessConfig "hinagata_app" database Nothing
migrate setupTarget = do
modifyIORef' migrationCalls (+ 1)
migrated <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString setupTarget)))
level <- queryText migrated "SHOW client_min_messages"
assert "setup role settings apply before migration" (level == "warning")
execCheck migrated "CREATE TABLE items (id integer PRIMARY KEY, note text NOT NULL)"
execCheck migrated "GRANT SELECT, INSERT ON items TO hinagata_app"
execCheck migrated "CREATE INDEX CONCURRENTLY items_note_idx ON items (note)"
PQ.finish migrated
pure (Right ())
verify setupTarget = do
verified <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString setupTarget)))
count <- queryInt verified "SELECT count(*) FROM items"
PQ.finish verified
pure (if count == 2 then Right () else Left "unexpected baseline rows")
baselineSpec = BaselineSpec lifecycleConfig goodPlan (Just "migration-v1") (Just "verify-v1") ["plpgsql"] (Just "C") migrate verify Nothing Nothing
firstBaseline <- ensureBaseline baselineSpec
secondBaseline <- ensureBaseline baselineSpec
(firstRef, firstReport) <- orFail firstBaseline
(secondRef, secondReport) <- orFail secondBaseline
calls <- readIORef migrationCalls
assert "baseline builds once and reuses a sealed generation" (baselineDatabase firstRef == baselineDatabase secondRef && kind firstReport == Built && kind secondReport == Reused && comparison firstReport == Nothing && comparison secondReport == Nothing && calls == 1)
recordedManifest <- queryText connection "SELECT fingerprint_manifest::text FROM hinagata_test.generations WHERE state = 'Ready' LIMIT 1"
assert "fingerprint manifest omits raw role-setting values" (not ("warning" `ByteString.isInfixOf` recordedManifest))
rotatedPassword <- orFail (mkSecret "rotated-passphrase")
let rotatedConfig = lifecycleConfig {administration = adminRole {password = Just rotatedPassword}, setup = setupRole {password = Just rotatedPassword}, application = appRole {password = Just rotatedPassword}}
rotated <- ensureBaseline baselineSpec {configuration = rotatedConfig, compareAgainst = Just firstRef}
assert "credential rotation does not invalidate a baseline" (case rotated of Right (ref, report) -> baselineDatabase ref == baselineDatabase firstRef && kind report == Reused && fmap changedComponents (comparison report) == Just []; _ -> False)
let sealedSql = Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> databaseNameText (baselineDatabase firstRef) <> "' AND datallowconn = false")
assert "ready baseline rejects connections" =<< ((== 1) <$> queryInt connection sealedSql)
createDirectoryIfMissing True (fixtureRoot </> "scenario")
ByteString.writeFile (fixtureRoot </> "scenario" </> "fixture.sql") "INSERT INTO items (id, note) VALUES (3, 'scenario');"
scenarioPlan <- bundleOrFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixtureName "good", fixtureName "scenario"]
let consume info = do
app <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString (applicationTarget info))))
count <- queryInt app "SELECT count(*) FROM items"
assert "clone contains one base prefix plus scenario suffix" (count == 3)
execCheck app "INSERT INTO items (id, note) VALUES (4, 'application')"
PQ.finish app
pure (connectionDatabase (applicationTarget info))
firstLease <- withDatabase lifecycleConfig firstRef scenarioPlan consume
secondLease <- withDatabase lifecycleConfig firstRef scenarioPlan consume
firstClone <- orFail firstLease
secondClone <- orFail secondLease
assert "bracketed clones are distinct" (firstClone /= secondClone)
assert "released clone databases are dropped" =<< ((== 0) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname IN ('" <> databaseNameText firstClone <> "', '" <> databaseNameText secondClone <> "')")))
assert "clone release records terminal state" =<< ((== 2) <$> queryInt connection "SELECT count(*) FROM hinagata_test.leases WHERE state = 'Released'")
firstStarted <- newEmptyMVar
secondStarted <- newEmptyMVar
releaseCallbacks <- newEmptyMVar
firstDone <- newEmptyMVar
secondDone <- newEmptyMVar
_ <- forkIO (try @SomeException (withDatabase lifecycleConfig firstRef scenarioPlan (\info -> consume info >>= \clone -> putMVar firstStarted clone >> takeMVar releaseCallbacks >> pure clone)) >>= putMVar firstDone)
_ <- forkIO (try @SomeException (withDatabase lifecycleConfig firstRef scenarioPlan (\info -> consume info >>= \clone -> putMVar secondStarted clone >> takeMVar releaseCallbacks >> pure clone)) >>= putMVar secondDone)
warmFirst <- timeout 10000000 (takeMVar firstStarted)
warmSecond <- timeout 10000000 (takeMVar secondStarted)
assert "concurrent warm callers hold distinct writable clones" (case (warmFirst, warmSecond) of (Just left, Just right) -> left /= right; _ -> False)
putMVar releaseCallbacks ()
putMVar releaseCallbacks ()
warmFirstDone <- timeout 10000000 (takeMVar firstDone)
warmSecondDone <- timeout 10000000 (takeMVar secondDone)
assert "concurrent callbacks release both clones" (case (warmFirstDone, warmSecondDone) of (Just (Right (Right _)), Just (Right (Right _))) -> True; _ -> False)
thrown <- try @SomeException (withDatabase lifecycleConfig firstRef scenarioPlan (\_ -> fail "intentional callback failure"))
assert "callback exception propagates after cleanup" (case thrown of Left _ -> True; Right _ -> False)
assert "exceptional callback released its clone" =<< ((== 5) <$> queryInt connection "SELECT count(*) FROM hinagata_test.leases WHERE state = 'Released'")
badScenario <- bundleOrFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixtureName "bad"]
failedLoad <- withDatabase lifecycleConfig firstRef badScenario (\_ -> pure ())
assert "failed scenario load refuses callback" (case failedLoad of Left _ -> True; Right _ -> False)
assert "failed scenario load releases its clone" =<< ((== 6) <$> queryInt connection "SELECT count(*) FROM hinagata_test.leases WHERE state = 'Released'")
assert "lease scenario preserves primary SQLSTATE" (case failedLoad of Left LeaseError {sqlState = Just "23505"} -> True; _ -> False)
directFailure <- withDatabase lifecycleConfig firstRef goodPlan (\LeaseInfo {applicationTarget} -> loadInto applicationTarget defaultSessionOptions badScenario)
assert "direct scenario observes the same constraint failure" (case directFailure of Right (Left LoadError {phase = ExecuteSql, sqlState = Just "23505"}) -> True; _ -> False)
assert "lease and direct load preserve matching structured context" (case (failedLoad, directFailure) of (Left LeaseError {fixtureLoadContext = Just context}, Right (Left direct)) -> context == loadErrorContext direct; _ -> False)
createDirectoryIfMissing True (fixtureRoot </> "slow-acquisition")
ByteString.writeFile (fixtureRoot </> "slow-acquisition" </> "fixture.sql") "SELECT pg_sleep(2);"
slowScenario <- bundleOrFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixtureName "good", fixtureName "slow-acquisition"]
callbackStarted <- newIORef False
releasedBeforeTimeout <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations WHERE state = 'Released'"
timedSetup <- withDatabase (lifecycleConfig {acquisitionDeadlineMs = positive 1500}) firstRef slowScenario (\_ -> modifyIORef' callbackStarted (const True))
releasedAfterTimeout <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations WHERE state = 'Released'"
ranTimedCallback <- readIORef callbackStarted
assert "one deadline stops slow scenario setup before callback and releases its clone" (case timedSetup of Left LeaseError {cause = "acquisition deadline expired"} -> not ranTimedCallback && releasedAfterTimeout == releasedBeforeTimeout + 1; _ -> False)
slowConsumer <- withDatabase (lifecycleConfig {acquisitionDeadlineMs = positive 1500}) firstRef scenarioPlan (\_ -> threadDelay 1700000 >> pure True)
assert "acquisition deadline stops at callback handoff" (slowConsumer == Right True)
timedLease <- withDatabaseClassified lifecycleConfig firstRef slowScenario ReleaseAlways (const False) (\_ -> pure (42 :: Int))
assert "lease timing separates scenario load from clone and release" (case timedLease of Right LeaseOutcome {callbackValue = 42, timings = LeaseTimings {queueWaitMs = 0, cloneMs, scenarioLoadMs, completionMs}} -> cloneMs > 0 && scenarioLoadMs >= 1500 && completionMs > 0; _ -> False)
cloneHookCalls <- newIORef (0 :: Int)
let prepareCloneHook setupTarget = do
modifyIORef' cloneHookCalls (+ 1)
prepared <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString setupTarget)))
execCheck prepared "CREATE TABLE clone_hook_probe (value integer NOT NULL)"
execCheck prepared "INSERT INTO clone_hook_probe (value) VALUES (42)"
execCheck prepared "GRANT SELECT ON clone_hook_probe TO hinagata_app"
PQ.finish prepared
pure (Right ())
hookedBaseline <- ensureBaseline baselineSpec {clonePreparationHook = Just prepareCloneHook}
(hookedRef, hookedReport) <- orFail hookedBaseline
assert "adding clone preparation does not rebuild an unchanged template" (kind hookedReport == Reused && baselineDatabase hookedRef == baselineDatabase firstRef)
let inspectHook info = do
app <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString (applicationTarget info))))
value <- queryInt app "SELECT value FROM clone_hook_probe"
PQ.finish app
pure value
firstHooked <- withDatabase lifecycleConfig hookedRef scenarioPlan inspectHook
secondHooked <- withDatabase lifecycleConfig hookedRef scenarioPlan inspectHook
hookCalls <- readIORef cloneHookCalls
assert "clone preparation runs separately before both callback handoffs" (firstHooked == Right 42 && secondHooked == Right 42 && hookCalls == 2)
failedHookBaseline <- ensureBaseline baselineSpec {clonePreparationHook = Just (\_ -> pure (Left "intentional hook failure"))}
(failedHookRef, _) <- orFail failedHookBaseline
failedHookCallback <- newIORef False
failedHookLease <- withDatabase lifecycleConfig failedHookRef scenarioPlan (\_ -> modifyIORef' failedHookCallback (const True))
calledAfterFailedHook <- readIORef failedHookCallback
assert "failed clone preparation releases its clone before callback" (case failedHookLease of Left LeaseError {cause = "clone preparation hook failed"} -> not calledAfterFailedHook; _ -> False)
withManager lifecycleConfig $ \manager -> do
TastyAdapter.runDatabaseCase manager firstRef scenarioPlan (\info -> consume info >> pure (Right ()))
let requests = DatabaseRequest "zeta" firstRef scenarioPlan :| [DatabaseRequest "alpha" firstRef scenarioPlan]
collection <-
withDatabases
manager
requests
( \entries -> do
assert "named collection orders acquisitions stably" (map fst entries == ["alpha", "zeta"])
traverse (consume . snd) entries
)
clones <- orFail collection
assert "named collection has simultaneous distinct clones" (case clones of [left, right] -> left /= right; _ -> False)
beforePartial <- queryInt connection "SELECT count(*) FROM hinagata_test.leases WHERE state = 'Released'"
partial <- withDatabases manager (DatabaseRequest "alpha" firstRef scenarioPlan :| [DatabaseRequest "zeta" firstRef badScenario]) (\_ -> pure ())
afterPartial <- queryInt connection "SELECT count(*) FROM hinagata_test.leases WHERE state = 'Released'"
assert "later collection failure unwinds earlier clones" (case partial of Left _ -> afterPartial == beforePartial + 2; Right _ -> False)
managedFailure <- withManagedDatabaseClassified manager firstRef scenarioPlan PreserveFailures not (\_ -> pure False)
managedInfo <- case managedFailure of
Right LeaseOutcome {callbackValue = False, leaseDisposition = LeasePreserved, leaseInfo} -> pure leaseInfo
_ -> fail "managed classified failure was not retained"
let LeaseInfo {leaseId = managedId} = managedInfo
managedRelease <- releaseLease lifecycleConfig managedId
assert "managed preserved lease releases by ID" (case managedRelease of Right (Released _) -> True; _ -> False)
managedDetached <- orFail =<< acquireManagedDetached manager firstRef scenarioPlan
let LeaseInfo {leaseId = managedDetachedId} = managedDetached
managedDetachedRelease <- releaseLease lifecycleConfig managedDetachedId
assert "manager capacity is released after detached acquisition" (case managedDetachedRelease of Right (Released _) -> True; _ -> False)
let tightConfig = lifecycleConfig {activeLeases = positive 1, pendingRequests = positive 1, setupWorkers = positive 1}
withManager tightConfig $ \manager -> do
allocationsBeforeOversize <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
oversize <- withDatabases manager (DatabaseRequest "a" firstRef scenarioPlan :| [DatabaseRequest "b" firstRef scenarioPlan]) (\_ -> pure ())
allocationsAfterOversize <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
assert "oversized collection is refused before allocation" (case oversize of Left _ -> allocationsBeforeOversize == allocationsAfterOversize; Right _ -> False)
held <- newEmptyMVar
releaseHeld <- newEmptyMVar
firstFinished <- newEmptyMVar
secondFinished <- newEmptyMVar
thirdFinished <- newEmptyMVar
_ <- forkIO (try @SomeException (withManagedDatabase manager firstRef scenarioPlan (\_ -> putMVar held () >> takeMVar releaseHeld)) >>= putMVar firstFinished)
_ <- takeMVar held
queuedWorker <- forkIO (try @SomeException (withManagedDatabase manager firstRef scenarioPlan (\_ -> pure ())) >>= putMVar secondFinished)
threadDelay 100000
saturated <- withManagedDatabase manager firstRef scenarioPlan (\_ -> pure ())
assert "manager rejects a full pending queue before allocation" (case saturated of Left _ -> True; Right _ -> False)
killThread queuedWorker
cancelledQueue <- timeout 1000000 (takeMVar secondFinished)
assert "queued admission can be cancelled" (case cancelledQueue of Just (Left _) -> True; _ -> False)
_ <- forkIO (try @SomeException (withManagedDatabase manager firstRef scenarioPlan (\_ -> pure ())) >>= putMVar thirdFinished)
putMVar releaseHeld ()
firstResult <- timeout 10000000 (takeMVar firstFinished)
thirdResult <- timeout 10000000 (takeMVar thirdFinished)
assert "manager releases capacity after callback and queue cancellation" (case (firstResult, thirdResult) of (Just (Right (Right _)), Just (Right (Right _))) -> True; _ -> False)
timingHeld <- newEmptyMVar
timingRelease <- newEmptyMVar
timingHolderDone <- newEmptyMVar
timingWaiterDone <- newEmptyMVar
_ <- forkIO (withManagedDatabase manager firstRef scenarioPlan (\_ -> putMVar timingHeld () >> takeMVar timingRelease) >>= putMVar timingHolderDone)
_ <- takeMVar timingHeld
_ <- forkIO (withManagedDatabaseClassified manager firstRef scenarioPlan ReleaseAlways (const False) (\_ -> pure ()) >>= putMVar timingWaiterDone)
threadDelay 200000
putMVar timingRelease ()
timingHolder <- timeout 10000000 (takeMVar timingHolderDone)
timingWaiter <- timeout 10000000 (takeMVar timingWaiterDone)
assert "managed timing records active-queue wait" (case (timingHolder, timingWaiter) of (Just (Right ()), Just (Right LeaseOutcome {timings = LeaseTimings {queueWaitMs}})) -> queueWaitMs >= 150; _ -> False)
let deadlineConfig = tightConfig {acquisitionDeadlineMs = positive 1000}
withManager deadlineConfig $ \manager -> do
held <- newEmptyMVar
releaseHeld <- newEmptyMVar
finished <- newEmptyMVar
_ <- forkIO (try @SomeException (withManagedDatabase manager firstRef scenarioPlan (\_ -> putMVar held () >> takeMVar releaseHeld)) >>= putMVar finished)
_ <- takeMVar held
expired <- withManagedDatabase manager firstRef scenarioPlan (\_ -> pure ())
assert "queued acquisition obeys its deadline" (case expired of Left _ -> True; Right _ -> False)
putMVar releaseHeld ()
holder <- timeout 10000000 (takeMVar finished)
assert "deadline does not cancel another caller's lease" (case holder of Just (Right (Right _)) -> True; _ -> False)
afterDeadline <- withManagedDatabase manager firstRef scenarioPlan (\_ -> pure ())
assert "deadline releases pending admission" (case afterDeadline of Right _ -> True; Left _ -> False)
connectionCeilingTest lifecycleConfig firstRef scenarioPlan connection
eightWarmTest lifecycleConfig firstRef scenarioPlan connection
processCrashTests lifecycleConfig baselineSpec connection fixtureRoot bundleRoot
cleanupLifecycleTests lifecycleConfig catalog firstRef scenarioPlan connection database
completionCancellationTests lifecycleConfig firstRef scenarioPlan connection target
retentionLifecycleTests lifecycleConfig firstRef scenarioPlan connection catalog
bulkRequested <- (== Just "1") <$> lookupEnv "HINAGATA_TEST_BULK_BASELINE"
when bulkRequested (bulkBaselineTests lifecycleConfig fixtureRoot bundleRoot connection)
missingCreatedbTests lifecycleConfig baselineSpec connection database
structuredLoadErrorTests lifecycleConfig baselineSpec goodPlan fixtureRoot bundleRoot connection
roleSettingTests lifecycleConfig baselineSpec scenarioPlan connection
generationLifecycleTests lifecycleConfig baselineSpec firstRef scenarioPlan fixtureRoot bundleRoot connection migrationCalls messageSetting target catalogSchema
missingCreatedbTests :: HinagataConfig -> BaselineSpec -> PQ.Connection -> DatabaseName -> IO ()
missingCreatedbTests configuration specification observer database = do
execCheck observer "CREATE ROLE hinagata_no_createdb LOGIN NOCREATEDB"
execCheck observer (Encoding.encodeUtf8 ("GRANT CONNECT, CREATE ON DATABASE \"" <> databaseNameText database <> "\" TO hinagata_no_createdb"))
schema <- orFail (mkSqlIdentifier "hinagata_no_createdb_test")
projectId <- orFail (mkProjectId "no-createdb-test")
let noCreatedbConfig = configuration {maintenanceSchema = schema, project = projectId, administration = AccessConfig "hinagata_no_createdb" database Nothing}
denied <- ensureBaseline specification {configuration = noCreatedbConfig, migrationRevision = Just "migration-no-createdb"}
assert "administration role without CREATEDB gets a typed permission diagnostic" (case denied of Left BaselineError {cause = "administration role lacks CREATEDB or template access", sqlState = Just "42501"} -> True; _ -> False)
assert "denied baseline remains Failed with no created database" =<< ((== 1) <$> queryInt observer "SELECT count(*) FROM hinagata_no_createdb_test.generations g LEFT JOIN pg_database d ON d.datname = g.database_name WHERE g.state = 'Failed' AND g.database_oid IS NULL AND d.oid IS NULL AND g.last_error = 'administration role lacks CREATEDB or template access'")
bulkBaselineTests :: HinagataConfig -> FilePath -> FilePath -> PQ.Connection -> IO ()
bulkBaselineTests configuration fixtureRoot bundleRoot observer = do
let directory = fixtureRoot </> "bulk-prepared"
csv = directory </> "rows.csv"
bundleConfig = BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)
rows = 1000000 :: Int
createDirectoryIfMissing True directory
ByteString.writeFile (directory </> "fixture.yaml") "name: bulk-prepared\nsteps:\n - sql: schema.sql\n - copy:\n table: {schema: public, name: bulk_prepared_items}\n columns: [id, payload]\n file: rows.csv\n format: csv\n header: false\n"
ByteString.writeFile (directory </> "schema.sql") "CREATE TABLE bulk_prepared_items (id integer PRIMARY KEY, payload text NOT NULL); GRANT SELECT ON bulk_prepared_items TO hinagata_app;"
withBinaryFile csv WriteMode $ \handle ->
forM_ [1, 1001 .. rows] $ \first ->
ByteString.hPut handle (ByteString.pack (concatMap (\number -> show number ++ ",payload\n") [first .. min rows (first + 999)]))
base <- bundleOrFail =<< compileFixtures bundleConfig [fixtureName "bulk-prepared"]
migrations <- newIORef (0 :: Int)
let migrate _ = modifyIORef' migrations (+ 1) >> pure (Right ())
verify setupTarget =
bracket
(PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString setupTarget))))
PQ.finish
( \session -> do
count <- queryInt session "SELECT count(*) FROM bulk_prepared_items"
pure (if count == rows then Right () else Left "bulk baseline row count differs from input")
)
specification = BaselineSpec configuration base (Just "bulk-migration-v1") (Just "bulk-verify-v1") ["plpgsql"] (Just "C") migrate verify Nothing Nothing
(baseline, firstReport) <- orFail =<< ensureBaseline specification
(reused, reuseReport) <- orFail =<< ensureBaseline specification
migrationCount <- readIORef migrations
assert "million-row baseline loads once and then reuses its sealed generation" (kind firstReport == Built && kind reuseReport == Reused && baselineDatabase baseline == baselineDatabase reused && migrationCount == 1)
let countClone reference plan = withDatabase configuration reference plan $ \LeaseInfo {applicationTarget} ->
bracket
(PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString applicationTarget))))
PQ.finish
(\session -> queryInt session "SELECT count(*) FROM bulk_prepared_items")
firstClone <- countClone baseline base
secondClone <- countClone baseline base
assert "repeated clones retain one million rows without replaying duplicate-key COPY" (firstClone == Right rows && secondClone == Right rows)
start <- newEmptyMVar
finished <- mapM (const newEmptyMVar) [1 .. 8 :: Int]
mapM_ (\slot -> forkIO (readMVar start >> acquireDetached configuration baseline base >>= putMVar slot)) finished
putMVar start ()
let observe peak 0 = pure peak
observe peak remaining = do
active <- queryInt observer "SELECT count(*) FROM pg_stat_activity WHERE state = 'active' AND query LIKE 'CREATE DATABASE %' AND pid <> pg_backend_pid()"
completed <- mapM tryReadMVar finished
let nextPeak = max peak active
if all isJust completed then pure nextPeak else threadDelay 10000 >> observe nextPeak (remaining - 1)
maxCreates <- observe 0 (3000 :: Int)
acquired <- mapM (timeout 30000000 . takeMVar) finished
infos <- traverse (maybe (fail "bulk clone acquisition timed out") orFail) acquired
let cloneNames = map (connectionDatabase . applicationTarget) infos
assert "eight bulk clones have distinct databases" (length (nub cloneNames) == 8)
assert "large-template CREATE DATABASE statements overlap" (maxCreates >= 2)
releases <- traverse (\LeaseInfo {leaseId} -> releaseLease configuration leaseId) infos
assert "eight bulk clones release after concurrency measurement" (all (\case Right (Released _) -> True; _ -> False) releases)
withBinaryFile csv ReadWriteMode $ \handle -> ByteString.hPut handle "1,changed\n"
changedPlan <- bundleOrFail =<< compileFixtures bundleConfig [fixtureName "bulk-prepared"]
(changed, changedReport) <- orFail =<< ensureBaseline specification {basePlan = changedPlan, compareAgainst = Just baseline}
changedMigrations <- readIORef migrations
assert "changed bulk CSV bytes select a distinct generation" (kind changedReport == Built && baselineDatabase changed /= baselineDatabase baseline && fmap changedComponents (comparison changedReport) == Just [BaseFixtures] && changedMigrations == 2)
changedValue <- withDatabase configuration changed changedPlan $ \LeaseInfo {applicationTarget} ->
bracket
(PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString applicationTarget))))
PQ.finish
(\session -> queryText session "SELECT payload FROM bulk_prepared_items WHERE id = 1")
assert "changed bulk generation includes the modified source byte" (changedValue == Right "changed")
putStrLn ("million-row prepared baseline, reuse, repeated clones, source-byte invalidation, and concurrent CREATE passed; max_active_create=" ++ show maxCreates)
connectionCeilingTest :: HinagataConfig -> BaselineRef -> FixturePlan -> PQ.Connection -> IO ()
connectionCeilingTest configuration baseline scenario connection = withManager configuration $ \manager -> do
ready <- mapM (const newEmptyMVar) [1 .. 4 :: Int]
finished <- mapM (const newEmptyMVar) [1 .. 4 :: Int]
release <- newEmptyMVar
let holdApplication info signal =
bracket
(PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString (applicationTarget info)))))
PQ.finish
(\app -> queryInt app "SELECT count(*) FROM items" >> putMVar signal () >> takeMVar release)
mapM_
(\(signal, done) -> forkIO (try @SomeException (withManagedDatabase manager baseline scenario (\info -> holdApplication info signal)) >>= putMVar done))
(zip ready finished)
arrivals <- mapM (timeout 10000000 . takeMVar) ready
assert "four managed callbacks reach their application sessions" (all (== Just ()) arrivals)
observed <- queryInt connection "SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend' AND usename IN (current_user, 'hinagata_app')"
assert "four active leases plus one application session each stay within nine client connections including the observer" (observed == 9)
mapM_ (const (putMVar release ())) [1 .. 4 :: Int]
outcomes <- mapM (timeout 10000000 . takeMVar) finished
assert "measured callbacks all release their managed leases" (all (\case Just (Right (Right ())) -> True; _ -> False) outcomes)
eightWarmTest :: HinagataConfig -> BaselineRef -> FixturePlan -> PQ.Connection -> IO ()
eightWarmTest configuration baseline scenario connection = do
let templateName = databaseNameText (baselineDatabase baseline)
generation <- queryText connection (Encoding.encodeUtf8 ("SELECT id FROM hinagata_test.generations WHERE database_name = '" <> templateName <> "'"))
let key = "generation:" <> generation
hold = queryInt connection ("SELECT 1 FROM (SELECT pg_advisory_lock_shared(1212761905, hashtext('" <> key <> "'))) held")
unlock = queryText connection ("SELECT pg_advisory_unlock_shared(1212761905, hashtext('" <> key <> "'))::text")
bracket hold (const unlock) $ \_ -> do
ready <- mapM (const newEmptyMVar) [1 .. 8 :: Int]
finished <- mapM (const newEmptyMVar) [1 .. 8 :: Int]
release <- newEmptyMVar
let consume number signal info =
bracket
(PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString (applicationTarget info)))))
PQ.finish
( \app -> do
before <- queryInt app "SELECT count(*) FROM items"
assert "each warm clone starts with one base and scenario closure" (before == 3)
execCheck app (ByteString.pack ("INSERT INTO items (id, note) VALUES (" ++ show (1000 + number) ++ ", 'private')"))
after <- queryInt app "SELECT count(*) FROM items"
assert "each warm clone accepts an isolated application write" (after == 4)
putMVar signal (connectionDatabase (applicationTarget info))
takeMVar release
)
mapM_
(\(number, signal, done) -> forkIO (try @SomeException (withDatabase configuration baseline scenario (consume number signal)) >>= putMVar done))
(zip3 [1 .. 8 :: Int] ready finished)
arrivals <- mapM (timeout 20000000 . takeMVar) ready
assert "eight warm callbacks acquire while a shared generation lock is held" (all isJust arrivals)
let names = catMaybes arrivals
assert "eight warm callbacks hold distinct writable databases" (length (nub names) == 8)
mapM_ (const (putMVar release ())) [1 .. 8 :: Int]
outcomes <- mapM (timeout 20000000 . takeMVar) finished
assert "all eight warm callbacks release cleanly" (all (\case Just (Right (Right ())) -> True; _ -> False) outcomes)
processCrashTests :: HinagataConfig -> BaselineSpec -> PQ.Connection -> FilePath -> FilePath -> IO ()
processCrashTests configuration baselineSpec connection fixtureRoot bundleRoot = do
let slowDirectory = fixtureRoot </> "crash-loading"
createDirectoryIfMissing True slowDirectory
ByteString.writeFile (slowDirectory </> "fixture.sql") "SELECT pg_sleep(30);"
executable <- getExecutablePath
runBuilderCase executable
runSealCase executable
runPublishedCase executable
runUnboundCase executable "intent" "LOCK TABLE pg_database IN SHARE ROW EXCLUSIVE MODE" "CREATE DATABASE %" False
-- PostgreSQL 18's CREATE DATABASE itself waits on pg_shdescription, so a
-- table lock cannot isolate the post-CREATE/pre-COMMENT gap. The binding
-- boundary below covers a created database with no catalog OID instead.
runBindingCase executable
runDropCase executable
mapM_ (runCase executable) ["loading", "handoff"]
where
runSealCase executable = do
let revision = "migration-seal-crash"
readyFile = bundleRoot </> "crash-sealing.ready"
command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, "sealing", readyFile]
generationSql prefix = Encoding.encodeUtf8 ("SELECT " <> prefix <> " FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%" <> revision <> "%' ORDER BY created_at DESC LIMIT 1")
crashedDatabase <-
bracket
( do
execCheck connection "CREATE FUNCTION hinagata_test.pause_generation_publication() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(30); RETURN NEW; END $$"
execCheck connection "CREATE TRIGGER pause_generation_publication BEFORE UPDATE ON hinagata_test.generations FOR EACH ROW WHEN (OLD.state = 'Building' AND NEW.state = 'Ready') EXECUTE FUNCTION hinagata_test.pause_generation_publication()"
)
( \_ -> do
execCheck connection "DROP TRIGGER pause_generation_publication ON hinagata_test.generations"
execCheck connection "DROP FUNCTION hinagata_test.pause_generation_publication()"
)
( \_ -> withCreateProcess command $ \_ _ _ child -> do
(database, serverPid) <- waitForChild child $ do
rows <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%" <> revision <> "%'"))
if rows == 0
then pure Nothing
else do
state <- queryText connection (generationSql "state")
database <- Encoding.decodeUtf8 <$> queryText connection (generationSql "database_name::text")
sealed <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "' AND NOT datallowconn"))
sleeping <- queryInt connection "SELECT count(*) FROM pg_stat_activity WHERE query LIKE 'UPDATE %generations%' AND wait_event = 'PgSleep' AND pid <> pg_backend_pid()"
if state == "Building" && sealed == 1 && sleeping == 1
then do
serverPid <- queryInt connection "SELECT pid FROM pg_stat_activity WHERE query LIKE 'UPDATE %generations%' AND wait_event = 'PgSleep' AND pid <> pg_backend_pid() LIMIT 1"
pure (Just (database, serverPid))
else pure Nothing
processId <- maybe (fail "sealing worker has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert "killed sealing worker exits abnormally" (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
activeBackend <- queryInt connection (ByteString.pack ("SELECT count(*) FROM pg_stat_activity WHERE pid = " ++ show serverPid))
when (activeBackend > 0) (void (queryText connection (ByteString.pack ("SELECT pg_terminate_backend(" ++ show serverPid ++ ")::text"))))
waitForBackendGone "sealing" serverPid
pure database
)
interrupted <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%" <> revision <> "%' AND state = 'Building'"))
assert "sealed generation remains unpublished after process death" (interrupted == 1)
rebuilt <- ensureBaseline baselineSpec {migrationRevision = Just revision}
assert "caller rebuilds after seal-to-publication crash" (case rebuilt of Right (_, report) -> kind report == Built; Left _ -> False)
failed <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%" <> revision <> "%' AND state = 'Failed'"))
ready <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%" <> revision <> "%' AND state = 'Ready'"))
oldSealed <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> crashedDatabase <> "' AND NOT datallowconn"))
assert "interrupted sealed database remains inspectable while one replacement is ready" (failed == 1 && ready == 1 && oldSealed == 1)
runPublishedCase executable = do
let revision = "migration-published-crash"
readyFile = bundleRoot </> "crash-published.ready"
command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, "published", readyFile]
withCreateProcess command $ \_ _ _ child -> do
database <- waitForChild child $ do
exists <- doesFileExist readyFile
if exists then Just . Encoding.decodeUtf8 <$> ByteString.readFile readyFile else pure Nothing
ready <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.generations WHERE database_name = '" <> database <> "' AND state = 'Ready'"))
sealed <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "' AND NOT datallowconn"))
assert "published worker left one sealed ready generation" (ready == 1 && sealed == 1)
processId <- maybe (fail "published worker has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert "killed published worker exits abnormally" (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
reused <- ensureBaseline baselineSpec {migrationRevision = Just revision}
assert "ready generation survives publisher process death and is reused" (case reused of Right (baseline, report) -> kind report == Reused && databaseNameText (baselineDatabase baseline) == database; Left _ -> False)
runBindingCase executable = do
before <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
let command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, "binding", bundleRoot </> "crash-binding.ready"]
bracket
( do
execCheck connection "CREATE FUNCTION hinagata_test.pause_clone_binding() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(30); RETURN NEW; END $$"
execCheck connection "CREATE TRIGGER pause_clone_binding BEFORE UPDATE ON hinagata_test.allocations FOR EACH ROW WHEN (OLD.state = 'Allocating' AND NEW.state = 'Loading') EXECUTE FUNCTION hinagata_test.pause_clone_binding()"
)
( \_ -> do
execCheck connection "DROP TRIGGER pause_clone_binding ON hinagata_test.allocations"
execCheck connection "DROP FUNCTION hinagata_test.pause_clone_binding()"
)
( \_ -> withCreateProcess command $ \_ _ _ child -> do
(identifier, database, serverPid) <- waitForChild child $ do
count <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
if count <= before
then pure Nothing
else do
identifierBytes <- queryText connection "SELECT id FROM hinagata_test.allocations ORDER BY created_at DESC LIMIT 1"
let identifier = Encoding.decodeUtf8 identifierBytes
allocationSql suffix = Encoding.encodeUtf8 ("SELECT " <> suffix <> " FROM hinagata_test.allocations WHERE id = '" <> identifier <> "'")
state <- queryText connection (allocationSql "state")
marker <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database d JOIN pg_shdescription s ON s.objoid = d.oid AND s.classoid = 'pg_database'::regclass JOIN hinagata_test.allocations a ON a.database_name = d.datname WHERE a.id = '" <> identifier <> "' AND s.description = 'hinagata:v1:' || (SELECT cluster_uuid::text FROM hinagata_test.meta) || ':' || a.ownership_token::text"))
if state /= "Allocating" || marker /= 1
then pure Nothing
else do
generation <- queryText connection (allocationSql "generation_id")
let key = "generation:" <> Encoding.decodeUtf8 generation
backendSql prefix = Encoding.encodeUtf8 ("SELECT " <> prefix <> " FROM pg_locks WHERE locktype = 'advisory' AND classid = 1212761905::oid AND objid = hashtext('" <> key <> "')::oid AND granted AND pid <> pg_backend_pid()")
holders <- queryInt connection (backendSql "count(*)")
if holders == 0
then pure Nothing
else do
serverPid <- queryInt connection (backendSql "pid" <> " LIMIT 1")
sleeping <- queryInt connection (ByteString.pack ("SELECT count(*) FROM pg_stat_activity WHERE pid = " ++ show serverPid ++ " AND wait_event = 'PgSleep'"))
if sleeping == 1
then do
database <- Encoding.decodeUtf8 <$> queryText connection (allocationSql "database_name::text")
pure (Just (identifier, database, serverPid))
else pure Nothing
processId <- maybe (fail "binding worker has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert "killed marker-binding worker exits abnormally" (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
activeBackend <- queryInt connection (ByteString.pack ("SELECT count(*) FROM pg_stat_activity WHERE pid = " ++ show serverPid))
when (activeBackend > 0) (void (queryText connection (ByteString.pack ("SELECT pg_terminate_backend(" ++ show serverPid ++ ")::text"))))
waitForBackendGone "binding" serverPid
candidates <- orFail =<< planCleanup configuration
candidate <- maybe (fail "marker-bound allocation is absent from cleanup preview") pure (find (\item -> allocationId item == identifier) candidates)
assert "marker exists but catalog identity remains unbound" (state candidate == "Allocating" && disposition candidate == Ambiguous)
assert "marker-binding crash leaves its OID unbound" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.allocations WHERE id = '" <> identifier <> "' AND database_oid IS NULL")))
assert "marker-bound database survives worker death" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "'")))
refused <- applyCleanup configuration OrphansOnly [identifier]
assert "cleanup refuses a marker-only allocation without an OID" (refused == Right [Refused identifier "allocation has no bound database OID"])
assert "refused marker-only cleanup leaves the database present" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "'")))
)
runDropCase executable = do
let readyFile = bundleRoot </> "crash-drop.ready"
command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, "drop", readyFile]
withCreateProcess command $ \_ _ _ child -> do
lease <- waitForChild child $ do
present <- doesFileExist readyFile
if present then Just . Encoding.decodeUtf8 <$> ByteString.readFile readyFile else pure Nothing
allocation <- Encoding.decodeUtf8 <$> queryText connection (Encoding.encodeUtf8 ("SELECT allocation_id FROM hinagata_test.leases WHERE id = '" <> lease <> "'"))
database <- Encoding.decodeUtf8 <$> queryText connection (Encoding.encodeUtf8 ("SELECT database_name::text FROM hinagata_test.allocations WHERE id = '" <> allocation <> "'"))
let leaseKey = "lease:" <> lease
serverPid <- queryInt connection (Encoding.encodeUtf8 ("SELECT pid FROM pg_locks WHERE locktype = 'advisory' AND classid = 1212761905::oid AND objid = hashtext('" <> leaseKey <> "')::oid AND granted AND pid <> pg_backend_pid()"))
let triggerName = "pause_drop_release"
activeSql = ByteString.pack ("SELECT count(*) FROM pg_stat_activity WHERE pid = " ++ show serverPid ++ " AND state = 'active'")
_ <-
bracket
( do
execCheck connection "CREATE FUNCTION hinagata_test.pause_drop_release() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(30); RETURN NEW; END $$"
execCheck connection (Encoding.encodeUtf8 ("CREATE TRIGGER " <> triggerName <> " BEFORE UPDATE ON hinagata_test.allocations FOR EACH ROW WHEN (NEW.id = '" <> allocation <> "' AND NEW.state = 'Released') EXECUTE FUNCTION hinagata_test.pause_drop_release()"))
)
( \_ -> do
execCheck connection (Encoding.encodeUtf8 ("DROP TRIGGER " <> triggerName <> " ON hinagata_test.allocations"))
execCheck connection "DROP FUNCTION hinagata_test.pause_drop_release()"
)
( \_ -> do
writeReady (readyFile ++ ".go") "go"
observed <- try @SomeException $ waitForChild child $ do
state <- queryText connection (Encoding.encodeUtf8 ("SELECT state FROM hinagata_test.allocations WHERE id = '" <> allocation <> "'"))
present <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "'"))
active <- queryInt connection activeSql
if state == "Releasing" && present == 0 && active > 0
then pure (Just ())
else pure Nothing
case observed of
Right () -> pure ()
Left exception -> do
state <- queryText connection (Encoding.encodeUtf8 ("SELECT state FROM hinagata_test.allocations WHERE id = '" <> allocation <> "'"))
present <- queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "'"))
active <- queryInt connection activeSql
fail ("drop boundary was not reached: " ++ show exception ++ "; state=" ++ show state ++ "; present=" ++ show present ++ "; matching-backends=" ++ show active)
processId <- maybe (fail "drop worker has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert "killed drop worker exits abnormally" (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
activeBackend <- queryInt connection (ByteString.pack ("SELECT count(*) FROM pg_stat_activity WHERE pid = " ++ show serverPid))
when (activeBackend > 0) (void (queryText connection (ByteString.pack ("SELECT pg_terminate_backend(" ++ show serverPid ++ ")::text"))))
waitForBackendGone "drop" serverPid
)
assert "killed drop worker leaves its database absent" =<< ((== 0) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> database <> "'")))
candidates <- orFail =<< planCleanup configuration
candidate <- maybe (fail "killed drop allocation is absent from cleanup preview") pure (find (\item -> allocationId item == allocation) candidates)
assert "killed drop worker leaves a Releasing allocation with a missing clone" (state candidate == "Releasing" && disposition candidate == Missing)
released <- applyCleanup configuration OrphansOnly [allocation]
assert "explicit cleanup idempotently records the interrupted drop" (released == Right [Released allocation])
runUnboundCase executable mode catalogLock activePattern expectDatabase = do
before <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
let command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, mode, bundleRoot </> ("crash-" ++ mode ++ ".ready")]
activeSql prefix = "SELECT " <> prefix <> " FROM pg_stat_activity WHERE state = 'active' AND query LIKE '" <> activePattern <> "' AND pid <> pg_backend_pid()"
(identifier, serverPid) <-
bracket
(execCheck connection "BEGIN" >> execCheck connection catalogLock)
(const (execCheck connection "ROLLBACK"))
( \_ -> withCreateProcess command $ \_ _ _ child -> do
boundary <- try @SomeException $ waitForChild child $ do
count <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
if count <= before
then pure Nothing
else do
latest <- queryText connection "SELECT id FROM hinagata_test.allocations ORDER BY created_at DESC LIMIT 1"
state <- queryText connection ("SELECT state FROM hinagata_test.allocations WHERE id = '" <> latest <> "'")
pending <- queryInt connection (activeSql "count(*)")
if state == "Allocating" && pending > 0
then do
backend <- queryInt connection (activeSql "pid" <> " LIMIT 1")
pure (Just (Encoding.decodeUtf8 latest, backend))
else pure Nothing
(identifier, serverPid) <- case boundary of
Right observed -> pure observed
Left failure -> do
activity <- queryText connection "SELECT coalesce(string_agg(pid::text || ':' || state || ':' || query, E'\\n'), '') FROM pg_stat_activity WHERE pid <> pg_backend_pid()"
fail ("unbound " ++ mode ++ " boundary failed: " ++ show failure ++ "; activity=" ++ show activity)
processId <- maybe (fail "unbound worker has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert ("killed " ++ mode ++ " worker exits abnormally") (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
pure (identifier, serverPid)
)
waitForBackendGone mode serverPid
candidates <- orFail =<< planCleanup configuration
candidate <- maybe (fail "unbound allocation is absent from cleanup preview") pure (find (\item -> allocationId item == identifier) candidates)
assert ("unbound " ++ mode ++ " allocation remains ambiguous after process death") (state candidate == "Allocating" && disposition candidate == Ambiguous)
let CleanupCandidate {database = intendedDatabase} = candidate
existenceSql = Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> databaseNameText intendedDatabase <> "'")
beforeApply <- queryInt connection existenceSql
when expectDatabase (assert "database creation completed before marker binding" (beforeApply == 1))
applied <- applyCleanup configuration OrphansOnly [identifier]
afterApply <- queryInt connection existenceSql
assert ("ambiguous " ++ mode ++ " allocation refuses deletion") (applied == Right [Refused identifier "allocation has no bound database OID"] && afterApply == beforeApply)
waitForBackendGone mode serverPid = loop (100 :: Int)
where
statement = ByteString.pack ("SELECT count(*) FROM pg_stat_activity WHERE pid = " ++ show serverPid)
loop 0 = fail ("server backend continued after killed " ++ mode ++ " worker")
loop remaining = do
active <- queryInt connection statement
if active == 0 then pure () else threadDelay 100000 >> loop (remaining - 1)
runBuilderCase executable = do
let readyFile = bundleRoot </> "crash-building.ready"
command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, "building", readyFile]
withCreateProcess command $ \_ _ _ child -> do
_ <- waitForChild child $ do
exists <- doesFileExist readyFile
if not exists
then pure Nothing
else do
bytes <- ByteString.readFile readyFile
pure (if bytes == "building" then Just () else Nothing)
building <- queryInt connection "SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%migration-process-crash%' AND state = 'Building'"
assert "process-kill builder reached a recorded Building generation" (building == 1)
processId <- maybe (fail "crash builder has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert "killed baseline builder exits abnormally" (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
rebuilt <- ensureBaseline baselineSpec {migrationRevision = Just "migration-process-crash"}
assert "surviving caller publishes after builder process death" (case rebuilt of Right (_, report) -> kind report == Built; Left _ -> False)
failed <- queryInt connection "SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%migration-process-crash%' AND state = 'Failed'"
ready <- queryInt connection "SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%migration-process-crash%' AND state = 'Ready'"
assert "builder process death leaves one Failed and one Ready generation" (failed == 1 && ready == 1)
runCase executable mode = do
before <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
let readyFile = bundleRoot </> ("crash-" ++ mode ++ ".ready")
command = proc executable ["--crash-lease-worker", fixtureRoot, bundleRoot, mode, readyFile]
withCreateProcess command $ \_ _ _ child -> do
identifier <- waitForChild child $ case mode of
"loading" -> do
count <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
if count <= before
then pure Nothing
else do
latest <- queryText connection "SELECT id FROM hinagata_test.allocations ORDER BY created_at DESC LIMIT 1"
state <- queryText connection ("SELECT state FROM hinagata_test.allocations WHERE id = '" <> latest <> "'")
pure (if state == "Loading" then Just (Encoding.decodeUtf8 latest) else Nothing)
"handoff" -> do
exists <- doesFileExist readyFile
if not exists
then pure Nothing
else do
lease <- ByteString.readFile readyFile
if ByteString.null lease
then pure Nothing
else Just . Encoding.decodeUtf8 <$> queryText connection ("SELECT allocation_id FROM hinagata_test.leases WHERE id = '" <> lease <> "'")
_ -> fail "unknown crash phase"
processId <- maybe (fail "crash worker has no process ID") pure =<< getPid child
signalProcess sigKILL processId
ended <- timeout 10000000 (waitForProcess child)
assert ("killed " ++ mode ++ " worker exits abnormally") (case ended of Just ExitSuccess -> False; Just (ExitFailure _) -> True; Nothing -> False)
candidate <- waitForOrphan identifier
assert ("killed " ++ mode ++ " worker leaves an inspectable orphan") (state candidate == if mode == "loading" then "Loading" else "Active")
released <- applyCleanup configuration OrphansOnly [identifier]
assert ("explicit apply releases killed " ++ mode ++ " worker's clone") (released == Right [Released identifier])
let CleanupCandidate {database = crashedDatabase} = candidate
assert ("killed " ++ mode ++ " worker's database is absent after apply") =<< ((== 0) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> databaseNameText crashedDatabase <> "'")))
waitForOrphan identifier = loop (100 :: Int)
where
loop 0 = fail "killed worker's allocation did not become orphaned"
loop remaining = do
candidates <- orFail =<< planCleanup configuration
case find (\candidate -> allocationId candidate == identifier && disposition candidate == Orphaned) candidates of
Just candidate -> pure candidate
Nothing -> threadDelay 100000 >> loop (remaining - 1)
waitForChild :: (HasCallStack) => ProcessHandle -> IO (Maybe a) -> IO a
waitForChild child probe = loop (100 :: Int)
where
loop 0 = fail ("crash worker did not reach its requested boundary: " ++ prettyCallStack callStack)
loop remaining = do
found <- probe
case found of
Just value -> pure value
Nothing -> do
exited <- getProcessExitCode child
case exited of
Just status -> fail ("crash worker exited before its boundary: " ++ show status)
Nothing -> threadDelay 100000 >> loop (remaining - 1)
cleanupLifecycleTests :: HinagataConfig -> CatalogIdentity -> BaselineRef -> FixturePlan -> PQ.Connection -> DatabaseName -> IO ()
cleanupLifecycleTests lifecycleConfig catalog firstRef scenarioPlan connection database = do
generationBytes <- queryText connection (Encoding.encodeUtf8 ("SELECT id FROM hinagata_test.generations WHERE database_name = '" <> databaseNameText (baselineDatabase firstRef) <> "'"))
let generationText = Encoding.decodeUtf8 generationBytes
orphanId = "11111111-1111-4111-8111-111111111111"
orphanLease = "22222222-2222-4222-8222-222222222222"
orphanDatabase = "hinagata_cleanup_orphan"
orphanToken = "33333333-3333-4333-8333-333333333333"
orphanMarker = "hinagata:v1:" <> clusterUuid catalog <> ":" <> orphanToken
execCheck connection (Encoding.encodeUtf8 ("CREATE DATABASE " <> orphanDatabase))
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE " <> orphanDatabase <> " IS '" <> orphanMarker <> "'"))
orphanOid <- queryInt connection (Encoding.encodeUtf8 ("SELECT oid FROM pg_database WHERE datname = '" <> orphanDatabase <> "'"))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.allocations (id, project_id, generation_id, run_id, database_name, database_oid, ownership_token, state) VALUES ('" <> orphanId <> "', 'baseline-test', '" <> generationText <> "', 'recovery-run', '" <> orphanDatabase <> "', " <> Text.pack (show orphanOid) <> "::oid, '" <> orphanToken <> "', 'Active')"))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.leases (id, allocation_id, run_id, state) VALUES ('" <> orphanLease <> "', '" <> orphanId <> "', 'recovery-run', 'Active')"))
candidates <- orFail =<< planCleanup lifecycleConfig
assert "free session lock classifies a positively owned active clone as orphaned" (case [disposition candidate | candidate <- candidates, allocationId candidate == orphanId] of [Orphaned] -> True; _ -> False)
assert "cleanup preview leaves an orphaned database present" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> orphanDatabase <> "'")))
applied <- applyCleanup lifecycleConfig OrphansOnly [orphanId]
assert "explicit apply removes a revalidated orphan" (applied == Right [Released orphanId])
assert "orphaned clone is absent after apply" =<< ((== 0) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> orphanDatabase <> "'")))
reapplied <- applyCleanup lifecycleConfig OrphansOnly [orphanId]
assert "recovery release is idempotent" (reapplied == Right [AlreadyReleased orphanId])
liveStarted <- newEmptyMVar
releaseLive <- newEmptyMVar
liveFinished <- newEmptyMVar
_ <- forkIO (try @SomeException (withDatabase lifecycleConfig firstRef scenarioPlan (\info -> putMVar liveStarted info >> takeMVar releaseLive)) >>= putMVar liveFinished)
liveInfo <- takeMVar liveStarted
livePreview <- orFail =<< planCleanup lifecycleConfig
let LeaseInfo {leaseId = liveLease} = liveInfo
liveCandidates = [candidate | candidate@CleanupCandidate {leaseId = Just identifier} <- livePreview, identifier == liveLease]
assert "held lease session lock appears live in cleanup preview" (case liveCandidates of [CleanupCandidate {disposition = Live}] -> True; _ -> False)
liveAllocation <- case liveCandidates of [candidate] -> pure (allocationId candidate); _ -> fail "live allocation is absent"
liveInspection <- inspectLease lifecycleConfig liveLease
assert "direct lease inspection reports a live callback" (case liveInspection of Right (Just CleanupCandidate {disposition = Live}) -> True; _ -> False)
livePreserve <- preserveLease lifecycleConfig liveLease
assert "explicit preservation skips a live callback" (livePreserve == Right (Skipped liveAllocation Live))
liveApply <- applyCleanup lifecycleConfig IncludeRetained [liveAllocation]
assert "explicit apply skips a live callback" (liveApply == Right [Skipped liveAllocation Live])
putMVar releaseLive ()
liveResult <- timeout 10000000 (takeMVar liveFinished)
assert "live callback releases normally after cleanup skips it" (case liveResult of Just (Right (Right _)) -> True; _ -> False)
lostOwner <- withDatabase lifecycleConfig firstRef scenarioPlan $ \info@LeaseInfo {leaseId = lostLease} -> do
let key = "lease:" <> lostLease
statement = Encoding.encodeUtf8 ("SELECT pg_terminate_backend(pid)::text FROM pg_locks WHERE locktype = 'advisory' AND classid = 1212761905::oid AND objid = hashtext('" <> key <> "')::oid AND granted AND pid <> pg_backend_pid()")
terminated <- queryText connection statement
assert "lost-owner test terminates the held maintenance lock" (terminated == "true")
pure info
assert "lease scope reports loss of its owning maintenance session" (case lostOwner of Left _ -> True; Right _ -> False)
lostCandidates <- orFail =<< planCleanup lifecycleConfig
lostCandidate <- case [candidate | candidate@CleanupCandidate {state = "Active", disposition = Orphaned} <- lostCandidates] of
[candidate] -> pure candidate
_ -> fail "lost-owner allocation was not classified orphaned"
recoveredOwner <- applyCleanup lifecycleConfig OrphansOnly [allocationId lostCandidate]
assert "lost-owner clone requires explicit ownership-checked release" (recoveredOwner == Right [Released (allocationId lostCandidate)])
interruptedLease <- newEmptyMVar
continuedAfterLoss <- newIORef False
lostDuringCallback <- timeout 10000000 $ withDatabase lifecycleConfig firstRef scenarioPlan $ \info@LeaseInfo {leaseId = lostLease} -> do
putMVar interruptedLease info
let key = "lease:" <> lostLease
statement = Encoding.encodeUtf8 ("SELECT pg_terminate_backend(pid)::text FROM pg_locks WHERE locktype = 'advisory' AND classid = 1212761905::oid AND objid = hashtext('" <> key <> "')::oid AND granted AND pid <> pg_backend_pid()")
terminated <- queryText connection statement
assert "running-callback test terminates the held maintenance lock" (terminated == "true")
threadDelay 30000000
modifyIORef' continuedAfterLoss (const True)
assert "ownership loss interrupts a still-running callback" (case lostDuringCallback of Just (Left LeaseError {cause = "lease ownership connection was lost during callback"}) -> True; _ -> False)
ranAfterLoss <- readIORef continuedAfterLoss
assert "interrupted callback does not continue after lease ownership loss" (not ranAfterLoss)
LeaseInfo {leaseId = interruptedId} <- takeMVar interruptedLease
interruptedInspection <- inspectLease lifecycleConfig interruptedId
interruptedAllocation <- case interruptedInspection of
Right (Just candidate@CleanupCandidate {state = "Active", disposition = Orphaned}) -> pure (allocationId candidate)
_ -> fail "interrupted callback did not leave an inspectable orphan"
interruptedRelease <- applyCleanup lifecycleConfig OrphansOnly [interruptedAllocation]
assert "interrupted callback requires explicit ownership-checked cleanup" (interruptedRelease == Right [Released interruptedAllocation])
let retainedId = "44444444-4444-4444-8444-444444444444"
retainedLease = "55555555-5555-4555-8555-555555555555"
retainedDatabase = "hinagata_cleanup_retained"
retainedToken = "66666666-6666-4666-8666-666666666666"
retainedMarker = "hinagata:v1:" <> clusterUuid catalog <> ":" <> retainedToken
execCheck connection (Encoding.encodeUtf8 ("CREATE DATABASE " <> retainedDatabase))
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE " <> retainedDatabase <> " IS '" <> retainedMarker <> "'"))
retainedOid <- queryInt connection (Encoding.encodeUtf8 ("SELECT oid FROM pg_database WHERE datname = '" <> retainedDatabase <> "'"))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.allocations (id, project_id, generation_id, run_id, database_name, database_oid, ownership_token, state) VALUES ('" <> retainedId <> "', 'baseline-test', '" <> generationText <> "', 'recovery-run', '" <> retainedDatabase <> "', " <> Text.pack (show retainedOid) <> "::oid, '" <> retainedToken <> "', 'Preserved')"))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.leases (id, allocation_id, run_id, state) VALUES ('" <> retainedLease <> "', '" <> retainedId <> "', 'recovery-run', 'Preserved')"))
retainedPreview <- orFail =<< planCleanup lifecycleConfig
assert "preserved lease is classified as retained" (case [disposition candidate | candidate <- retainedPreview, allocationId candidate == retainedId] of [Retained] -> True; _ -> False)
ordinary <- applyCleanup lifecycleConfig OrphansOnly [retainedId]
assert "ordinary recovery skips a preserved database" (ordinary == Right [Skipped retainedId Retained])
selected <- applyCleanup lifecycleConfig IncludeRetained [retainedId]
assert "explicit retained selection releases a positively owned database" (selected == Right [Released retainedId])
let foreignId = "77777777-7777-4777-8777-777777777777"
foreignLease = "88888888-8888-4888-8888-888888888888"
foreignDatabase = "hinagata_cleanup_foreign"
foreignToken = "99999999-9999-4999-8999-999999999999"
execCheck connection (Encoding.encodeUtf8 ("CREATE DATABASE " <> foreignDatabase))
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE " <> foreignDatabase <> " IS 'foreign'"))
foreignOid <- queryInt connection (Encoding.encodeUtf8 ("SELECT oid FROM pg_database WHERE datname = '" <> foreignDatabase <> "'"))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.allocations (id, project_id, generation_id, run_id, database_name, database_oid, ownership_token, state) VALUES ('" <> foreignId <> "', 'baseline-test', '" <> generationText <> "', 'recovery-run', '" <> foreignDatabase <> "', " <> Text.pack (show foreignOid) <> "::oid, '" <> foreignToken <> "', 'Active')"))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.leases (id, allocation_id, run_id, state) VALUES ('" <> foreignLease <> "', '" <> foreignId <> "', 'recovery-run', 'Active')"))
foreignPreview <- orFail =<< planCleanup lifecycleConfig
assert "altered marker is visible as foreign evidence" (case [disposition candidate | candidate <- foreignPreview, allocationId candidate == foreignId] of [Foreign] -> True; _ -> False)
foreignApply <- applyCleanup lifecycleConfig IncludeRetained [foreignId]
assert "apply refuses a same-prefix database with altered ownership marker" (case foreignApply of Right [Refused identifier _] -> identifier == foreignId; _ -> False)
assert "refused foreign database remains present" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> foreignDatabase <> "'")))
execCheck connection (Encoding.encodeUtf8 ("DROP DATABASE " <> foreignDatabase))
let ambiguousId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
ambiguousDatabase = "hinagata_cleanup_ambiguous"
ambiguousToken = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"
execCheck connection (Encoding.encodeUtf8 ("CREATE DATABASE " <> ambiguousDatabase))
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.allocations (id, project_id, generation_id, run_id, database_name, ownership_token, state) VALUES ('" <> ambiguousId <> "', 'baseline-test', '" <> generationText <> "', 'recovery-run', '" <> ambiguousDatabase <> "', '" <> ambiguousToken <> "', 'Allocating')"))
ambiguousPreview <- orFail =<< planCleanup lifecycleConfig
assert "unbound allocation remains ambiguous even when its name exists" (case [disposition candidate | candidate <- ambiguousPreview, allocationId candidate == ambiguousId] of [Ambiguous] -> True; _ -> False)
ambiguousApply <- applyCleanup lifecycleConfig IncludeRetained [ambiguousId]
assert "apply refuses an unbound database identity" (case ambiguousApply of Right [Refused identifier _] -> identifier == ambiguousId; _ -> False)
assert "ambiguous same-prefix database remains present" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> ambiguousDatabase <> "'")))
execCheck connection (Encoding.encodeUtf8 ("DROP DATABASE " <> ambiguousDatabase))
let protectedId = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"
execCheck connection (Encoding.encodeUtf8 ("INSERT INTO hinagata_test.allocations (id, project_id, generation_id, run_id, database_name, ownership_token, state) VALUES ('" <> protectedId <> "', 'baseline-test', '" <> generationText <> "', 'recovery-run', '" <> databaseNameText database <> "', '" <> ambiguousToken <> "', 'Allocating')"))
protectedApply <- applyCleanup lifecycleConfig IncludeRetained [protectedId]
assert "maintenance database is protected even with a catalog allocation" (case protectedApply of Right [Refused identifier _] -> identifier == protectedId; _ -> False)
structuredLoadErrorTests :: HinagataConfig -> BaselineSpec -> FixturePlan -> FilePath -> FilePath -> PQ.Connection -> IO ()
structuredLoadErrorTests configuration specification goodPlan fixtureRoot bundleRoot observer = do
project <- orFail (mkProjectId "structured-error-test")
secret <- orFail (mkSecret "fixture-password-secret")
let selected = configuration {project, setup = (setup configuration) {password = Just secret}, application = (application configuration) {password = Just secret}}
migrate target = do
original <- migrationHook specification target
case original of
Left problem -> pure (Left problem)
Right () -> bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))) PQ.finish $ \connection -> do
execCheck connection "CREATE TABLE secret_items (id text PRIMARY KEY)"
execCheck connection "CREATE TABLE parent_items (id integer PRIMARY KEY)"
execCheck connection "CREATE TABLE child_items (parent_id integer REFERENCES parent_items(id) DEFERRABLE INITIALLY DEFERRED)"
execCheck connection "GRANT SELECT, INSERT ON secret_items, child_items TO hinagata_app"
pure (Right ())
spec = specification {configuration = selected, migrationRevision = Just "structured-errors-v1", migrationHook = migrate}
config = BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)
sqlDirectory = fixtureRoot </> "error-sql"
copyDirectory = fixtureRoot </> "error-copy"
terminateDirectory = fixtureRoot </> "error-terminate"
mapM_ (createDirectoryIfMissing True) [sqlDirectory, copyDirectory, terminateDirectory]
ByteString.writeFile (sqlDirectory </> "fixture.yaml") "name: error-sql\nsteps:\n - sql: first.sql\n - sql: fail.sql\n"
ByteString.writeFile (sqlDirectory </> "first.sql") "INSERT INTO secret_items VALUES ('fixture-value-secret');"
ByteString.writeFile (sqlDirectory </> "fail.sql") "INSERT INTO secret_items VALUES ('fixture-value-secret');"
ByteString.writeFile (copyDirectory </> "fixture.yaml") "name: error-copy\nsteps:\n - sql: first.sql\n - copy:\n table: {schema: public, name: secret_items}\n columns: [id]\n file: rows.csv\n format: csv\n header: false\n"
ByteString.writeFile (copyDirectory </> "first.sql") "INSERT INTO items VALUES (13, 'early write');"
ByteString.writeFile (copyDirectory </> "rows.csv") "fixture-copy-secret\nfixture-copy-secret\n"
ByteString.writeFile (terminateDirectory </> "fixture.sql") "SELECT pg_terminate_backend(pg_backend_pid());"
(baseline, _) <- orFail =<< ensureBaseline spec
withManager selected $ \manager -> forM_ [("error-sql", ExecuteSql, Just 1, "23505"), ("error-copy", ExecuteCopy, Just 1, "23505"), ("deferred", CommitLoad, Nothing, "23503")] $ \(name, expectedPhase, expectedStep, expectedState) -> do
plan <- bundleOrFail =<< compileFixtures config [fixtureName name]
combined <- bundleOrFail =<< compileFixtures config [fixtureName "good", fixtureName name]
direct <- orFail =<< withDatabase selected baseline goodPlan (\LeaseInfo {applicationTarget} -> loadInto applicationTarget defaultSessionOptions plan)
loadFailure <- case direct of Left problem -> pure problem; Right _ -> fail "expected direct fixture failure"
let expected = loadErrorContext loadFailure
matches context = context == Just expected
assert "direct load attributes SQL/COPY/COMMIT errors" (case expected of FixtureLoadContext {phase, fixture, stepIndex, sqlState} -> phase == expectedPhase && stepIndex == expectedStep && sqlState == Just expectedState && fixture == (if expectedPhase == CommitLoad then Nothing else Just (fixtureName name)))
baseFailure <- ensureBaseline spec {basePlan = combined}
assert "baseline retains structured fixture failure" (case baseFailure of Left BaselineError {sqlState, fixtureLoadContext} -> sqlState == Just expectedState && matches fixtureLoadContext; _ -> False)
leaseFailure <- withDatabaseClassified selected baseline plan ReleaseAlways (const False) (const (fail "failed scenario reached callback"))
managedFailure <- withManagedDatabaseClassified manager baseline plan ReleaseAlways (const False) (const (fail "failed managed scenario reached callback"))
assert "lease retains structured fixture failure" (case leaseFailure of Left LeaseError {sqlState, fixtureLoadContext} -> sqlState == Just expectedState && matches fixtureLoadContext; _ -> False)
assert "manager forwards original fixture context" (case managedFailure of Left LeaseError {sqlState, fixtureLoadContext} -> sqlState == Just expectedState && matches fixtureLoadContext; _ -> False)
let safe = Text.pack (show (fmap (const ()) baseFailure, fmap (const ()) leaseFailure, fmap (const ()) managedFailure))
assert "lifecycle Show omits fixture data credentials and endpoints" (all (not . (`Text.isInfixOf` safe)) ["fixture-value-secret", "fixture-copy-secret", "fixture-password-secret", "ConnectionTarget"])
preparedTarget <- newEmptyMVar
let hook target = bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))) PQ.finish $ \connection -> do
execCheck connection "BEGIN; INSERT INTO secret_items VALUES ('prepared'); PREPARE TRANSACTION 'hinagata_error_cleanup'"
putMVar preparedTarget target
pure (Right ())
(hooked, _) <- orFail =<< ensureBaseline spec {clonePreparationHook = Just hook}
terminating <- bundleOrFail =<< compileFixtures config [fixtureName "error-terminate"]
directTermination <- orFail =<< withDatabase selected baseline goodPlan (\LeaseInfo {applicationTarget} -> loadInto applicationTarget defaultSessionOptions terminating)
originalContext <- case directTermination of Left problem -> pure (loadErrorContext problem); Right _ -> fail "expected terminated direct load"
doubleFailure <- withDatabaseClassified selected hooked terminating ReleaseAlways (const False) (const (fail "terminated load reached callback"))
assert "direct disconnect records rollback failure" (case originalContext of FixtureLoadContext {phase = ExecuteSql, rollbackFailure = Just _} -> True; _ -> False)
-- Disconnect timing can change whether libpq receives the server SQLSTATE.
-- Check that release cannot replace the actual primary state with DROP's state.
assert ("rollback and clone release failures keep primary load context: " ++ show (fmap (const ()) doubleFailure)) (case doubleFailure of Left LeaseError {cause = "scenario fixture load failed", sqlState, fixtureLoadContext = Just FixtureLoadContext {phase = ExecuteSql, fixture = Just name, stepIndex = Just 0, sqlState = loadState, rollbackFailure = Just _}, cleanupFailure = Just _} -> name == fixtureName "error-terminate" && sqlState == loadState && sqlState /= Just "55006"; _ -> False)
target <- takeMVar preparedTarget
bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))) PQ.finish $ \connection -> execCheck connection "ROLLBACK PREPARED 'hinagata_error_cleanup'"
identifier <- queryText observer (Encoding.encodeUtf8 ("SELECT l.id FROM hinagata_test.leases l JOIN hinagata_test.allocations a ON a.id=l.allocation_id WHERE a.database_name='" <> databaseNameText (connectionDatabase target) <> "'"))
released <- orFail =<< releaseLease selected (Encoding.decodeUtf8 identifier)
assert "double failure stays recoverable through explicit release" (case released of Released _ -> True; _ -> False)
roleSettingTests :: HinagataConfig -> BaselineSpec -> FixturePlan -> PQ.Connection -> IO ()
roleSettingTests configuration specification scenario observer =
bracket_
(execCheck observer "ALTER ROLE hinagata_app SET standard_conforming_strings=off")
(execCheck observer "ALTER ROLE hinagata_app RESET standard_conforming_strings")
$ do
searchPath <- orFail (mkSqlIdentifier "search_path")
custom <- orFail (mkSqlIdentifier "hinagata.setting_probe")
readOnly <- orFail (mkSqlIdentifier "default_transaction_read_only")
transactionReadOnly <- orFail (mkSqlIdentifier "transaction_read_only")
messages <- orFail (mkSqlIdentifier "client_min_messages")
execCheck observer "GRANT SET ON PARAMETER \"hinagata.setting_probe\" TO hinagata_app"
project <- orFail (mkProjectId "setting-semantics-test")
let customValue = "apostrophe' backslash\\; SELECT secret_setting_probe;"
appPath = "\"$user\", \"comma,schema\", public, pg_catalog"
selected =
configuration
{ project,
setupSettings = [RoleSetting messages "warning", RoleSetting searchPath "public, pg_catalog"],
applicationSettings = [RoleSetting searchPath appPath, RoleSetting custom customValue, RoleSetting readOnly "on", RoleSetting transactionReadOnly "on", RoleSetting messages "notice"]
}
migrate target = bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))) PQ.finish $ \connection -> do
schemas <- queryText connection "SELECT array_to_string(current_schemas(false), ',')"
assert "setup search_path is a schema list before migration" (schemas == "public,pg_catalog")
execCheck connection "CREATE TABLE items (id integer PRIMARY KEY, note text NOT NULL)"
execCheck connection "GRANT SELECT, INSERT ON items TO hinagata_app"
execCheck connection "CREATE SCHEMA \"comma,schema\""
execCheck connection "CREATE TABLE \"comma,schema\".setting_relation (value integer)"
execCheck connection "INSERT INTO \"comma,schema\".setting_relation VALUES (42)"
execCheck connection "GRANT USAGE ON SCHEMA \"comma,schema\" TO hinagata_app"
execCheck connection "GRANT SELECT ON \"comma,schema\".setting_relation TO hinagata_app"
pure (Right ())
spec = specification {configuration = selected, migrationHook = migrate, verificationHook = const (pure (Right ()))}
consume LeaseInfo {applicationTarget} = bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString applicationTarget)))) PQ.finish $ \connection -> do
path <- queryText connection "SHOW search_path"
assert "application path preserves user token and quoted comma schema" (path == Encoding.encodeUtf8 appPath)
assert "quoted comma schema resolves as one name" =<< ((== 42) <$> queryInt connection "SELECT value FROM setting_relation")
assert "custom setting preserves apostrophe and backslash as data" =<< ((== Encoding.encodeUtf8 customValue) <$> queryText connection "SHOW hinagata.setting_probe")
assert "default read-only setting persists without blocking its own ALTER" =<< ((== "on") <$> queryText connection "SHOW default_transaction_read_only")
assert "scalar setting retains meaning" =<< ((== "notice") <$> queryText connection "SHOW client_min_messages")
(baseline, built) <- orFail =<< ensureBaseline spec
assert "setting semantics baseline built" (kind built == Built)
(_, warm) <- orFail =<< ensureBaseline spec
assert "corrected settings baseline reuses" (kind warm == Reused)
void (orFail =<< withDatabase selected baseline scenario consume)
let database = databaseNameText (baselineDatabase baseline)
selector = " WHERE database_name='" <> database <> "'"
legacyManifest = "replace(fingerprint_manifest->>'manifest', 'hinagata-fingerprint-v2', 'hinagata-fingerprint-v1')"
assert "corrected fingerprints have v2 marker" =<< ((== 1) <$> queryInt observer (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.generations" <> selector <> " AND fingerprint_manifest->>'manifest' LIKE 'hinagata-fingerprint-v2%'")))
-- Exercise supported old-manifest comparison through an opaque public handle;
-- keep its selected digest stable while changing only the stored format marker.
execCheck observer (Encoding.encodeUtf8 ("UPDATE hinagata_test.generations SET fingerprint_manifest=jsonb_set(fingerprint_manifest, '{manifest}', to_jsonb(" <> legacyManifest <> "))" <> selector))
(_, compared) <- orFail =<< ensureBaseline spec {compareAgainst = Just baseline}
assert "v1 comparison identifies only fingerprint version" (fmap changedComponents (comparison compared) == Just [FingerprintVersion])
-- Now give that synthetic legacy generation its genuine v1 manifest digest.
execCheck observer (Encoding.encodeUtf8 ("UPDATE hinagata_test.generations SET fingerprint=encode(sha256(convert_to(fingerprint_manifest->>'manifest','UTF8')), 'hex')" <> selector))
(corrected, rebuilt) <- orFail =<< ensureBaseline spec
assert "legacy semantics generation is not reused" (kind rebuilt == Built && baselineDatabase corrected /= baselineDatabase baseline)
(_, reused) <- orFail =<< ensureBaseline spec
assert "new semantics reuse after legacy invalidation" (kind reused == Reused)
assert "legacy baseline stays sealed and inspectable" =<< ((== 1) <$> queryInt observer (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname='" <> database <> "' AND NOT datallowconn")))
void (orFail =<< withDatabase selected corrected scenario consume)
invalid <- orFail (mkSqlIdentifier "hinagata_unknown_parameter")
let invalidSpec = spec {configuration = selected {applicationSettings = [RoleSetting custom "half-persisted-probe", RoleSetting invalid "bad"]}}
failed <- ensureBaseline invalidSpec
assert "invalid role setting fails preparation" (case failed of Left _ -> True; _ -> False)
assert "failed role-setting transaction leaves no partial setting" =<< ((== 0) <$> queryInt observer "SELECT count(*) FROM pg_db_role_setting WHERE setconfig @> ARRAY['hinagata.setting_probe=half-persisted-probe']")
completionCancellationTests :: HinagataConfig -> BaselineRef -> FixturePlan -> PQ.Connection -> ConnectionTarget -> IO ()
completionCancellationTests configuration baseline scenario observer target =
withManager configuration {activeLeases = positive 1, setupWorkers = positive 1} $ \manager ->
forM_ [ReleaseAlways, DetachAlways, PreserveFailures] $ \policy -> do
infoReady <- newEmptyMVar
allowReturn <- newEmptyMVar
finished <- newEmptyMVar
worker <-
forkIO $
try @SomeException
(withManagedDatabaseClassified manager baseline scenario policy (const True) (\info -> putMVar infoReady info >> takeMVar allowReturn))
>>= putMVar finished
LeaseInfo {leaseId} <- takeMVar infoReady
bracket (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))) PQ.finish $ \controller ->
bracket (execCheck controller "BEGIN") (const (execCheck controller "ROLLBACK")) $ \_ -> do
_ <- queryText controller (Encoding.encodeUtf8 ("SELECT a.id::text FROM hinagata_test.allocations a JOIN hinagata_test.leases l ON l.allocation_id=a.id WHERE l.id='" <> leaseId <> "' FOR UPDATE OF a"))
putMVar allowReturn ()
let waitBlocked attempts = do
count <- queryInt observer "SELECT count(*) FROM pg_stat_activity WHERE wait_event_type='Lock' AND query LIKE 'UPDATE %allocations%' AND pid <> pg_backend_pid()"
if count > 0 then pure () else if attempts <= (0 :: Int) then fail "completion never reached blocked update" else threadDelay 10000 >> waitBlocked (attempts - 1)
waitBlocked 500
killThread worker
result <- timeout 3000000 (takeMVar finished)
assert "completion cancellation preserves ThreadKilled" (case result of Just (Left exception) -> fromException exception == Just ThreadKilled; _ -> False)
inspected <- orFail =<< inspectLease configuration leaseId
assert "interrupted completion leaves recoverable orphan" (case inspected of Just CleanupCandidate {disposition = Orphaned} -> True; _ -> False)
released <- orFail =<< releaseLease configuration leaseId
assert "interrupted completion recovers with ownership checks" (case released of Released _ -> True; _ -> False)
reused <- withManagedDatabase manager baseline scenario (const (pure ()))
assert "completion cancellation returns manager permits" (reused == Right ())
retentionLifecycleTests :: HinagataConfig -> BaselineRef -> FixturePlan -> PQ.Connection -> CatalogIdentity -> IO ()
retentionLifecycleTests lifecycleConfig firstRef scenarioPlan connection catalog = do
classifiedSuccess <- withDatabaseClassified lifecycleConfig firstRef scenarioPlan PreserveFailures not (\_ -> pure True)
assert "successful classified value is returned unchanged and released" (case classifiedSuccess of Right LeaseOutcome {callbackValue = True, leaseDisposition = LeaseReleased, cleanupDiagnostic = Nothing} -> True; _ -> False)
classifiedFailure <- withDatabaseClassified lifecycleConfig firstRef scenarioPlan PreserveFailures not (\_ -> pure False)
preservedInfo <- case classifiedFailure of
Right LeaseOutcome {callbackValue = False, leaseDisposition = LeasePreserved, cleanupDiagnostic = Nothing, leaseInfo} -> pure leaseInfo
_ -> fail "classified failure did not preserve its clone"
preservedConnection <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString (applicationTarget preservedInfo))))
preservedRows <- queryInt preservedConnection "SELECT count(*) FROM items"
PQ.finish preservedConnection
assert "preserved callback value leaves its scenario rows inspectable" (preservedRows == 3)
let LeaseInfo {leaseId = preservedId} = preservedInfo
preservedRelease <- releaseLease lifecycleConfig preservedId
assert "classified failure remains explicitly releasable by lease ID" (case preservedRelease of Right (Released _) -> True; _ -> False)
detachedInfo <- orFail =<< acquireDetached lifecycleConfig firstRef scenarioPlan
let LeaseInfo {leaseId = detachedId} = detachedInfo
detachedPreview <- orFail =<< planCleanup lifecycleConfig
assert "detached acquisition is retained without an active callback slot" (case [disposition candidate | candidate@CleanupCandidate {leaseId = Just identifier} <- detachedPreview, identifier == detachedId] of [Retained] -> True; _ -> False)
detachedInspection <- inspectLease lifecycleConfig detachedId
assert "direct inspection reports a detached lease" (case detachedInspection of Right (Just CleanupCandidate {state = "Detached", disposition = Retained}) -> True; _ -> False)
let detachedDatabase = databaseNameText (connectionDatabase (applicationTarget detachedInfo))
detachedToken <- queryText connection (Encoding.encodeUtf8 ("SELECT a.ownership_token::text FROM hinagata_test.allocations a JOIN hinagata_test.leases l ON l.allocation_id = a.id WHERE l.id = '" <> detachedId <> "'"))
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE \"" <> detachedDatabase <> "\" IS 'foreign'"))
foreignInspection <- inspectLease lifecycleConfig detachedId
assert "inspection exposes changed ownership evidence on a retained clone" (case foreignInspection of Right (Just CleanupCandidate {disposition = Foreign}) -> True; _ -> False)
foreignPreserve <- preserveLease lifecycleConfig detachedId
assert "explicit preservation refuses changed ownership evidence" (case foreignPreserve of Right (Refused _ _) -> True; _ -> False)
let detachedMarker = "hinagata:v1:" <> clusterUuid catalog <> ":" <> Encoding.decodeUtf8 detachedToken
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE \"" <> detachedDatabase <> "\" IS '" <> detachedMarker <> "'"))
detachedPreserve <- preserveLease lifecycleConfig detachedId
assert "detached lease can be explicitly preserved" (case detachedPreserve of Right (Preserved _) -> True; _ -> False)
detachedRePreserve <- preserveLease lifecycleConfig detachedId
assert "explicit lease preservation is idempotent" (case detachedRePreserve of Right (Preserved _) -> True; _ -> False)
preservedInspection <- inspectLease lifecycleConfig detachedId
assert "explicit preservation updates the lease state" (case preservedInspection of Right (Just CleanupCandidate {state = "Preserved", disposition = Retained}) -> True; _ -> False)
detachedRelease <- releaseLease lifecycleConfig detachedId
assert "detached lease releases by explicit ID" (case detachedRelease of Right (Released _) -> True; _ -> False)
releasedInspection <- inspectLease lifecycleConfig detachedId
assert "released lease remains inspectable by ID" (case releasedInspection of Right (Just CleanupCandidate {state = "Released", disposition = Missing}) -> True; _ -> False)
detachedReRelease <- releaseLease lifecycleConfig detachedId
assert "detached release by lease ID is idempotent" (case detachedReRelease of Right (AlreadyReleased _) -> True; _ -> False)
dropBlockedInfo <- orFail =<< acquireDetached lifecycleConfig firstRef scenarioPlan
let LeaseInfo {leaseId = dropBlockedId, applicationTarget = dropBlockedTarget} = dropBlockedInfo
preparedGid = "hinagata-drop-" <> dropBlockedId
preparedConnectionString = Encoding.encodeUtf8 (connectionStringText (renderConnectionString dropBlockedTarget))
dropBlockedDatabase = databaseNameText (connectionDatabase dropBlockedTarget)
bracket (PQ.connectdb preparedConnectionString) PQ.finish $ \preparedConnection -> do
execCheck preparedConnection "BEGIN"
execCheck preparedConnection "INSERT INTO items (id, note) VALUES (123456, 'prepared drop blocker')"
execCheck preparedConnection (Encoding.encodeUtf8 ("PREPARE TRANSACTION '" <> preparedGid <> "'"))
blockedRelease <- releaseLease lifecycleConfig dropBlockedId
assert "prepared transaction makes force drop fail" (case blockedRelease of Left _ -> True; _ -> False)
assert "failed drop retains its database" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> dropBlockedDatabase <> "'")))
assert "failed drop records state and diagnostic" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.allocations a JOIN hinagata_test.leases l ON l.allocation_id = a.id WHERE l.id = '" <> dropBlockedId <> "' AND a.state = 'CleanupFailed' AND l.state = 'CleanupFailed' AND length(a.last_error) > 0")))
bracket (PQ.connectdb preparedConnectionString) PQ.finish $ \preparedConnection ->
execCheck preparedConnection (Encoding.encodeUtf8 ("ROLLBACK PREPARED '" <> preparedGid <> "'"))
retriedRelease <- releaseLease lifecycleConfig dropBlockedId
assert "failed drop can be retried after its blocker is cleared" (case retriedRelease of Right (Released _) -> True; _ -> False)
callbackBlocked <- withDatabaseClassified lifecycleConfig firstRef scenarioPlan ReleaseAlways (const False) $ \LeaseInfo {leaseId, applicationTarget} -> do
let connectionString = Encoding.encodeUtf8 (connectionStringText (renderConnectionString applicationTarget))
callbackGid = "hinagata-callback-drop-" <> leaseId
bracket (PQ.connectdb connectionString) PQ.finish $ \preparedConnection -> do
execCheck preparedConnection "BEGIN"
execCheck preparedConnection "INSERT INTO items (id, note) VALUES (123457, 'callback drop blocker')"
execCheck preparedConnection (Encoding.encodeUtf8 ("PREPARE TRANSACTION '" <> callbackGid <> "'"))
callbackBlockedInfo <- case callbackBlocked of
Right LeaseOutcome {leaseInfo, leaseDisposition = LeaseCleanupFailed, cleanupDiagnostic = Just _} -> pure leaseInfo
_ -> fail "callback force-drop failure was not reported separately from its callback result"
let LeaseInfo {leaseId = callbackBlockedId, applicationTarget = callbackBlockedTarget} = callbackBlockedInfo
callbackConnectionString = Encoding.encodeUtf8 (connectionStringText (renderConnectionString callbackBlockedTarget))
callbackPreparedGid = "hinagata-callback-drop-" <> callbackBlockedId
assert "callback force-drop failure retains its diagnostic" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.allocations a JOIN hinagata_test.leases l ON l.allocation_id = a.id WHERE l.id = '" <> callbackBlockedId <> "' AND a.state = 'CleanupFailed' AND l.state = 'CleanupFailed' AND length(a.last_error) > 0")))
bracket (PQ.connectdb callbackConnectionString) PQ.finish $ \preparedConnection ->
execCheck preparedConnection (Encoding.encodeUtf8 ("ROLLBACK PREPARED '" <> callbackPreparedGid <> "'"))
callbackRetried <- releaseLease lifecycleConfig callbackBlockedId
assert "callback force-drop failure can be retried by lease ID" (case callbackRetried of Right (Released _) -> True; _ -> False)
assert "successful callback cleanup retry clears the diagnostic" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.allocations a JOIN hinagata_test.leases l ON l.allocation_id = a.id WHERE l.id = '" <> callbackBlockedId <> "' AND a.state = 'Released' AND a.last_error IS NULL")))
thrownInfo <- newEmptyMVar
classifiedThrown <- try @SomeException (withDatabaseClassified lifecycleConfig firstRef scenarioPlan PreserveFailures (const False) (\info -> putMVar thrownInfo info >> fail "intentional classified exception"))
assert "classified callback exception keeps its original exception" (case classifiedThrown of Left _ -> True; Right _ -> False)
LeaseInfo {leaseId = thrownId} <- takeMVar thrownInfo
thrownPreview <- orFail =<< planCleanup lifecycleConfig
assert "thrown callback preserves its clone under policy" (case [disposition candidate | candidate@CleanupCandidate {leaseId = Just identifier} <- thrownPreview, identifier == thrownId] of [Retained] -> True; _ -> False)
thrownRelease <- releaseLease lifecycleConfig thrownId
assert "exception-preserved lease releases by explicit ID" (case thrownRelease of Right (Released _) -> True; _ -> False)
cancelledInfo <- newEmptyMVar
cancelledDone <- newEmptyMVar
cancelledThread <-
forkIO
( try @SomeException
(withDatabaseClassified lifecycleConfig firstRef scenarioPlan PreserveFailures (const False) (\info -> putMVar cancelledInfo info >> threadDelay 30000000 >> pure ()))
>>= putMVar cancelledDone
)
LeaseInfo {leaseId = cancelledId, applicationTarget = cancelledTarget} <- takeMVar cancelledInfo
killThread cancelledThread
cancelledResult <- timeout 10000000 (takeMVar cancelledDone)
assert "callback cancellation rethrows its asynchronous exception" (case cancelledResult of Just (Left _) -> True; _ -> False)
assert "callback cancellation releases despite failure-preservation policy" =<< ((== 1) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM hinagata_test.leases WHERE id = '" <> cancelledId <> "' AND state = 'Released'")))
assert "callback cancellation drops its database" =<< ((== 0) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> databaseNameText (connectionDatabase cancelledTarget) <> "'")))
generationLifecycleTests :: HinagataConfig -> BaselineSpec -> BaselineRef -> FixturePlan -> FilePath -> FilePath -> PQ.Connection -> IORef Int -> SqlIdentifier -> ConnectionTarget -> SqlIdentifier -> IO ()
generationLifecycleTests lifecycleConfig baselineSpec firstRef scenarioPlan fixtureRoot bundleRoot connection migrationCalls messageSetting target catalogSchema = do
startingCalls <- readIORef migrationCalls
changed <- ensureBaseline baselineSpec {migrationRevision = Just "migration-v2", compareAgainst = Just firstRef}
(changedRef, changedReport) <- orFail changed
changedCalls <- readIORef migrationCalls
assert "migration revision creates a distinct baseline with an explained change" (baselineDatabase changedRef /= baselineDatabase firstRef && kind changedReport == Built && fmap changedComponents (comparison changedReport) == Just [MigrationRevision] && changedCalls == startingCalls + 1)
ByteString.writeFile (fixtureRoot </> "good" </> "fixture.sql") "INSERT INTO items (id, note) VALUES (1, 'changed first'); INSERT INTO items (id, note) VALUES (2, 'second');"
changedPlan <- bundleOrFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixtureName "good"]
allocationsBeforeConflict <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
conflicting <- withDatabase lifecycleConfig firstRef changedPlan (\_ -> pure ())
allocationsAfterConflict <- queryInt connection "SELECT count(*) FROM hinagata_test.allocations"
assert "conflicting captured fixture refused before allocation" (case conflicting of Left _ -> allocationsBeforeConflict == allocationsAfterConflict; Right _ -> False)
ByteString.writeFile (fixtureRoot </> "good" </> "fixture.sql") "INSERT INTO items (id, note) VALUES (1, 'first'); INSERT INTO items (id, note) VALUES (2, 'second');"
changedFixture <- ensureBaseline baselineSpec {basePlan = changedPlan, compareAgainst = Just firstRef}
assert "base fixture bytes select a new generation with an explained change" (case changedFixture of Right (ref, report) -> baselineDatabase ref /= baselineDatabase firstRef && kind report == Built && fmap changedComponents (comparison report) == Just [BaseFixtures]; _ -> False)
changedHook <- ensureBaseline baselineSpec {verificationRevision = Just "verify-v2", compareAgainst = Just firstRef}
assert "verification hook revision is independently explained" (case changedHook of Right (_, report) -> kind report == Built && fmap changedComponents (comparison report) == Just [VerificationRevision]; _ -> False)
let changedRoleConfig = lifecycleConfig {applicationSettings = [RoleSetting messageSetting "notice"]}
changedRole <- ensureBaseline baselineSpec {configuration = changedRoleConfig, compareAgainst = Just firstRef}
assert "application role setting is independently explained without its value" (case changedRole of Right (_, report) -> kind report == Built && fmap changedComponents (comparison report) == Just [ApplicationRoleSettings]; _ -> False)
failed <- ensureBaseline baselineSpec {migrationRevision = Just "migration-v3", verificationHook = \_ -> pure (Left "verification failed")}
assert "failed rebuild remains unpublished" (case failed of Left _ -> True; Right _ -> False)
missingExtension <- ensureBaseline baselineSpec {migrationRevision = Just "migration-v4", requiredExtensions = ["not_installed"]}
assert "missing declared extension prevents publication" (case missingExtension of Left _ -> True; Right _ -> False)
unknownFirst <- ensureBaseline baselineSpec {migrationRevision = Nothing}
unknownSecond <- ensureBaseline baselineSpec {migrationRevision = Nothing}
assert "unknown migration revision disables persistent reuse" (case (unknownFirst, unknownSecond) of (Right (left, leftReport), Right (right, rightReport)) -> baselineDatabase left /= baselineDatabase right && kind leftReport == Built && kind rightReport == Built; _ -> False)
oldReady <- ensureBaseline baselineSpec
assert "failed rebuild preserves earlier ready baseline" (case oldReady of Right (ref, report) -> baselineDatabase ref == baselineDatabase firstRef && kind report == Reused; _ -> False)
firstCold <- newEmptyMVar
secondCold <- newEmptyMVar
let concurrentSpec = baselineSpec {migrationRevision = Just "migration-concurrent"}
_ <- forkIO (ensureBaseline concurrentSpec >>= putMVar firstCold)
_ <- forkIO (ensureBaseline concurrentSpec >>= putMVar secondCold)
concurrentFirst <- timeout 10000000 (takeMVar firstCold)
concurrentSecond <- timeout 10000000 (takeMVar secondCold)
assert "concurrent cold callers publish one generation" (case (concurrentFirst, concurrentSecond) of (Just (Right (left, leftReport)), Just (Right (right, rightReport))) -> baselineDatabase left == baselineDatabase right && [kind leftReport, kind rightReport] `elem` [[Built, Reused], [Reused, Built]]; _ -> False)
builderEntered <- newEmptyMVar
builderBlock <- newEmptyMVar
builderDone <- newEmptyMVar
let deathSpec = baselineSpec {migrationRevision = Just "migration-builder-death"}
blockedBuilder = deathSpec {migrationHook = \_ -> putMVar builderEntered () >> takeMVar builderBlock >> pure (Right ())}
builderThread <- forkIO (try @SomeException (ensureBaseline blockedBuilder) >>= putMVar builderDone)
builderStarted <- timeout 10000000 (takeMVar builderEntered)
assert "builder reaches migration after recording a generation" (case builderStarted of Just () -> True; _ -> False)
cancelledWaiterDone <- newEmptyMVar
cancelledWaiter <- forkIO (try @SomeException (ensureBaseline deathSpec) >>= putMVar cancelledWaiterDone)
threadDelay 100000
killThread cancelledWaiter
cancelledWaiterResult <- timeout 10000000 (takeMVar cancelledWaiterDone)
assert "cancelling one baseline waiter does not block indefinitely" (case cancelledWaiterResult of Just (Left _) -> True; _ -> False)
expiredWaiter <- timeout 10000000 (ensureBaseline (deathSpec {configuration = lifecycleConfig {setupDeadlineMs = positive 200}}))
assert "a waiter deadline expires while the builder still holds its lock" (case expiredWaiter of Just (Left _) -> True; _ -> False)
survivingWaiterDone <- newEmptyMVar
_ <- forkIO (ensureBaseline deathSpec >>= putMVar survivingWaiterDone)
threadDelay 100000
killThread builderThread
builderResult <- timeout 10000000 (takeMVar builderDone)
assert "builder process death releases its coordination session" (case builderResult of Just (Left _) -> True; _ -> False)
survivor <- timeout 10000000 (takeMVar survivingWaiterDone)
assert "a later waiter builds a ready generation after builder death" (case survivor of Just (Right (_, report)) -> kind report == Built; _ -> False)
interruptedRecords <- queryInt connection "SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%migration-builder-death%' AND state = 'Failed'"
survivingRecords <- queryInt connection "SELECT count(*) FROM hinagata_test.generations WHERE fingerprint_manifest->>'manifest' LIKE '%migration-builder-death%' AND state = 'Ready'"
assert "interrupted build becomes Failed while survivor is ready" (interruptedRecords == 1 && survivingRecords == 1)
retired <- retireBaseline lifecycleConfig changedRef
assert "sealed baseline retires under positive ownership" (retired == Right ())
assert "retired template database is absent" =<< ((== 0) <$> queryInt connection (Encoding.encodeUtf8 ("SELECT count(*) FROM pg_database WHERE datname = '" <> databaseNameText (baselineDatabase changedRef) <> "'")))
retiredRecord <- queryText connection (Encoding.encodeUtf8 ("SELECT state FROM hinagata_test.generations WHERE database_name = '" <> databaseNameText (baselineDatabase changedRef) <> "'"))
assert "retired generation remains as a catalog tombstone" (retiredRecord == "Retiring")
retiredAgain <- retireBaseline lifecycleConfig changedRef
assert "retirement is idempotent after the owned template disappears" (retiredAgain == Right ())
staleClone <- withDatabase lifecycleConfig changedRef scenarioPlan (\_ -> pure ())
assert "retired baseline cannot allocate another clone" (case staleClone of Left _ -> True; _ -> False)
replacement <- ensureBaseline baselineSpec {migrationRevision = Just "migration-v2"}
assert "retirement permits a new ready generation for the same fingerprint" (case replacement of Right (ref, report) -> kind report == Built && baselineDatabase ref /= baselineDatabase changedRef; _ -> False)
let tamperedName = databaseNameText (baselineDatabase firstRef)
execCheck connection (Encoding.encodeUtf8 ("COMMENT ON DATABASE \"" <> tamperedName <> "\" IS 'foreign'"))
tampered <- ensureBaseline baselineSpec
assert "ready generation with altered marker is refused" (case tampered of Left _ -> True; Right _ -> False)
execCheck connection "DROP TABLE hinagata_test.leases"
incompleteCatalog <- ensureCatalog target catalogSchema defaultSessionOptions
assert "incomplete version-1 catalog is refused" (case incompleteCatalog of Left _ -> True; _ -> False)
requiredEnv :: String -> IO String
requiredEnv name = maybe (fail (name ++ " is missing")) pure =<< lookupEnv name
execCheck :: PQ.Connection -> ByteString.ByteString -> IO ()
execCheck connection statement = do
result <- PQ.exec connection statement
case result of
Nothing -> fail "PostgreSQL returned no result"
Just value -> do
status <- PQ.resultStatus value
unless (status == PQ.CommandOk) (fail ("unexpected result: " ++ show status))
queryInt :: PQ.Connection -> ByteString.ByteString -> IO Int
queryInt connection statement = do
read . ByteString.unpack <$> queryText connection statement
queryText :: PQ.Connection -> ByteString.ByteString -> IO ByteString.ByteString
queryText connection statement = do
result <- PQ.exec connection statement
case result of
Nothing -> fail "PostgreSQL returned no result"
Just value -> do
cell <- PQ.getvalue value (PQ.toRow (0 :: Int)) (PQ.toColumn (0 :: Int))
maybe (fail "PostgreSQL returned null") pure cell
fixtureName :: Text -> FixtureName
fixtureName = either (error . Text.unpack) id . mkFixtureName
positive :: Int -> Positive
positive = either (error . Text.unpack) id . mkPositive "test limit"
bundleOrFail :: Either BundleError FixturePlan -> IO FixturePlan
bundleOrFail = either (fail . show) pure
orFail :: (Show e) => Either e a -> IO a
orFail = either (fail . show) pure
assert :: String -> Bool -> IO ()
assert label condition = unless condition $ putStrLn ("FAIL: " ++ label) >> exitFailure
waitForNoLoader :: PQ.Connection -> Int -> IO Bool
waitForNoLoader connection remaining = do
active <- queryInt connection "SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() AND pid <> pg_backend_pid()"
if active == 0
then pure True
else
if remaining <= 0
then pure False
else threadDelay 50000 >> waitForNoLoader connection (remaining - 50000)