packages feed

hinagata-postgres (empty) → 0.1.0.0

raw patch · 26 files changed

+5172/−0 lines, 26 filesdep +aesondep +asyncdep +base

Dependencies added: aeson, async, base, bytestring, cryptohash-sha256, directory, filepath, hinagata-core, hinagata-postgres, postgresql-libpq, process, tasty, tasty-hunit, temporary, text, time, unix

Files

+ CHANGELOG.md view
@@ -0,0 +1,23 @@+# Changelog++## 0.1.0.0 - 2026-10-18++### New Features++- Load verified SQL and streaming CSV fixtures in one transaction on an exclusive native PostgreSQL session.+- Build, verify, seal, reuse, compare, and retire baseline generations through a versioned maintenance catalog.+- Lease isolated clones with preparation hooks, scenario loading, classified callback outcomes, detached or retained leases, and lifecycle timing reports.+- Bound manager admission and named clone collections; inspect, preview, and apply ownership-checked cleanup with crash recovery.+- Upgrade owned maintenance catalogs from format 1 to format 2.++### Bug Fixes++- Enforce acquisition deadlines, cancel native commands, close interrupted resources, and interrupt callbacks after lease ownership is lost.+- Preserve PostgreSQL list and scalar role-setting semantics and invalidate baselines created with the earlier encoding.+- Preserve structured fixture context through baseline and lease errors, with independent rollback and release diagnostics.+- Recover interrupted baseline publication and clone release while refusing cleanup when ownership is ambiguous.++### Other Changes++- Include a test adapter example, concurrency benchmarks, crash-boundary tests, and packaged SQL catalog migrations.+- Add package changelog, repository metadata, required autogenerated-module declarations, and internal PVP dependency bounds.
+ LICENSE view
@@ -0,0 +1,24 @@+Copyright (c) 2026 Shinzui++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice,+   this list of conditions and the following disclaimer.+2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.+3. Neither the name of the copyright holder nor the names of its contributors+   may be used to endorse or promote products derived from this software+   without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ bench/Concurrency.hs view
@@ -0,0 +1,107 @@+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Exception (bracket)+import Control.Monad (forM)+import Data.Aeson qualified as Json+import Data.ByteString.Lazy.Char8 qualified as Lazy+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Database.PostgreSQL.LibPQ qualified as PQ+import GHC.Clock (getMonotonicTimeNSec)+import Hinagata.Config+import Hinagata.Connection+import Hinagata.Fixture.Bundle+import Hinagata.Fixture.Types (mkFixtureName)+import Hinagata.Postgres.Baseline+import Hinagata.Postgres.Lease+import Hinagata.Postgres.Manager+import Hinagata.Prelude+import Hinagata.Types+import System.Environment (getArgs, getEnv)++main :: IO ()+main = do+  count <-+    getArgs >>= \case+      [value] -> pure (read value :: Int)+      _ -> fail "usage: hinagata-postgres-concurrent-leases COUNT"+  unless (count `elem` [1, 4, 8]) (fail "COUNT must be 1, 4, or 8")+  socket <- Text.pack <$> getEnv "HINAGATA_TEST_PGHOST"+  portText <- getEnv "HINAGATA_TEST_PGPORT"+  user <- Text.pack <$> getEnv "HINAGATA_TEST_PGUSER"+  databaseText <- Text.pack <$> getEnv "HINAGATA_TEST_PGDATABASE"+  fixtureRoot <- getEnv "BENCH_FIXTURE_ROOT"+  bundleRoot <- getEnv "BENCH_BUNDLE_ROOT"+  host <- orFail (socketDirectory socket)+  port <- orFail (mkPort (read portText))+  database <- orFail (mkDatabaseName databaseText)+  project <- orFail (mkProjectId "benchmark-concurrency")+  catalogSchema <- orFail (mkSqlIdentifier "hinagata")+  publicSchema <- orFail (mkSqlIdentifier "public")+  fixture <- orFail (mkFixtureName "bulk-base")+  let role = AccessConfig user database Nothing+      configuration =+        HinagataConfig+          { endpoint = EndpointConfig host port,+            project,+            fixtureRoot,+            bundleRoot,+            maintenanceDatabase = database,+            maintenanceSchema = catalogSchema,+            administration = role,+            setup = role,+            application = role,+            setupGrants = [GrantConnect, GrantCreate, GrantTemporary],+            applicationGrants = [GrantConnect],+            setupSchemaGrants = [SchemaGrant publicSchema SchemaUsage, SchemaGrant publicSchema SchemaCreate],+            applicationSchemaGrants = [SchemaGrant publicSchema SchemaUsage],+            setupSettings = [],+            applicationSettings = [],+            cloneStrategy = WalLog,+            acquisitionDeadlineMs = positive 300000,+            setupDeadlineMs = positive 300000,+            setupWorkers = positive 4,+            activeLeases = positive 8,+            pendingRequests = positive 16,+            chunkSize = positive 65536,+            sqlSizeLimit = positive 1048576+          }+  plan <- orFail =<< compileFixtures (BundleConfig fixtureRoot bundleRoot (positive 1048576) (positive 65536)) [fixture]+  let noop _ = pure (Right ())+      specification = BaselineSpec configuration plan (Just "benchmark-noop-v1") (Just "benchmark-noop-v1") [] Nothing noop noop Nothing Nothing+  (baseline, _) <- orFail =<< ensureBaseline specification+  started <- getMonotonicTimeNSec+  outcomes <- withManager configuration $ \manager ->+    Async.mapConcurrently+      ( \_ -> do+          requestStart <- getMonotonicTimeNSec+          result <- withManagedDatabaseClassified manager baseline plan ReleaseAlways (const False) $ \info -> do+            callbackStart <- getMonotonicTimeNSec+            bracket+              (PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString (applicationTarget info)))))+              PQ.finish+              ( \connection -> do+                  status <- PQ.status connection+                  unless (status == PQ.ConnectionOk) (fail "application pool connection failed")+                  threadDelay 50000+              )+            pure callbackStart+          requestEnd <- getMonotonicTimeNSec+          pure (requestStart, requestEnd, result)+      )+      [1 .. count]+  ended <- getMonotonicTimeNSec+  timings <- forM outcomes $ \(requestStart, requestEnd, result) -> case result of+    Left problem -> fail ("lease failed: " ++ show problem)+    Right LeaseOutcome {callbackValue, leaseDisposition = LeaseReleased, cleanupDiagnostic = Nothing, timings = LeaseTimings {queueWaitMs, catalogMs, generationLockMs, cloneMs, scenarioLoadMs, completionMs}} ->+      pure $ Json.object ["requestMs" Json..= ((requestEnd - requestStart) `div` 1000000), "callbackStartNs" Json..= callbackValue, "queueWaitMs" Json..= queueWaitMs, "catalogMs" Json..= catalogMs, "generationLockMs" Json..= generationLockMs, "cloneMs" Json..= cloneMs, "scenarioLoadMs" Json..= scenarioLoadMs, "completionMs" Json..= completionMs]+    Right other -> fail ("lease cleanup failed: " ++ show other)+  Lazy.putStrLn (Json.encode (Json.object ["concurrency" Json..= count, "elapsedMs" Json..= ((ended - started) `div` 1000000), "timings" Json..= timings]))++positive :: Int -> Positive+positive = either (error . Text.unpack) id . mkPositive "benchmark limit"++orFail :: (Show e) => Either e a -> IO a+orFail = either (fail . show) pure
+ bench/Main.hs view
@@ -0,0 +1,85 @@+module Main (main) where++import Control.Monad (forM_)+import Data.ByteString.Char8 qualified as ByteString+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Database.PostgreSQL.LibPQ qualified as PQ+import GHC.Clock (getMonotonicTimeNSec)+import Hinagata.Connection+import Hinagata.Fixture.Bundle+import Hinagata.Fixture.Types+import Hinagata.Postgres.Load+import Hinagata.Postgres.Session+import Hinagata.Prelude+import Hinagata.Types+import System.Directory (createDirectoryIfMissing)+import System.Environment (getArgs, getEnv)+import System.FilePath ((</>))+import System.IO (IOMode (WriteMode), withBinaryFile)+import System.IO.Temp (withSystemTempDirectory)++main :: IO ()+main = do+  rows <-+    getArgs >>= \case+      [value] -> pure (read value)+      _ -> fail "usage: hinagata-postgres-direct-load ROWS"+  socket <- getEnv "HINAGATA_TEST_PGHOST"+  portText <- getEnv "HINAGATA_TEST_PGPORT"+  user <- getEnv "HINAGATA_TEST_PGUSER"+  databaseText <- getEnv "HINAGATA_TEST_PGDATABASE"+  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)+  withSystemTempDirectory "hinagata-load-bench-" $ \workspace -> do+    let fixtureRoot = workspace </> "fixtures"+        fixtureDirectory = fixtureRoot </> "bulk"+        tableName = "hinagata_bench_" ++ show rows+        config = BundleConfig fixtureRoot (workspace </> "bundles") (positive 1048576) (positive 65536)+    createDirectoryIfMissing True fixtureDirectory+    ByteString.writeFile (fixtureDirectory </> "fixture.yaml") (ByteString.pack ("name: bulk\nsteps:\n  - sql: schema.sql\n  - copy:\n      table: {schema: public, name: " ++ tableName ++ "}\n      columns: [id, payload]\n      file: rows.csv\n      format: csv\n      header: false\n"))+    ByteString.writeFile (fixtureDirectory </> "schema.sql") (ByteString.pack ("CREATE TABLE " ++ tableName ++ " (id integer PRIMARY KEY, payload text NOT NULL);"))+    generateStart <- getMonotonicTimeNSec+    withBinaryFile (fixtureDirectory </> "rows.csv") WriteMode $ \handle ->+      forM_ [1, 1001 .. rows] $ \first ->+        ByteString.hPut handle (ByteString.pack (concatMap (\n -> show n ++ ",payload\n") [first .. min rows (first + 999)]))+    generateEnd <- getMonotonicTimeNSec+    plan <- orFail =<< compileFixtures config [fixtureName "bulk"]+    compileEnd <- getMonotonicTimeNSec+    report <- orFail =<< loadInto target defaultSessionOptions plan+    loadEnd <- getMonotonicTimeNSec+    connection <- PQ.connectdb (Encoding.encodeUtf8 (connectionStringText (renderConnectionString target)))+    found <- queryCount connection tableName+    PQ.finish connection+    unless (found == rows) (fail "loaded row count differs from input")+    putStrLn $+      unwords+        [ "rows=" ++ show rows,+          "generate_ms=" ++ show ((generateEnd - generateStart) `div` 1000000),+          "compile_ms=" ++ show ((compileEnd - generateEnd) `div` 1000000),+          "load_ms=" ++ show ((loadEnd - compileEnd) `div` 1000000),+          "copy_ms=" ++ show (copyMs report),+          "verify_ms=" ++ show (verifyMs report),+          "bytes=" ++ show (bytesSent report),+          "rows_per_second=" ++ show (if elapsedMs report == 0 then 0 else fromIntegral rows * 1000 `div` fromIntegral (elapsedMs report) :: Int)+        ]++queryCount :: PQ.Connection -> String -> IO Int+queryCount connection tableName = do+  result <- PQ.exec connection (ByteString.pack ("SELECT count(*) FROM " ++ tableName))+  case result of+    Nothing -> fail "count query returned no result"+    Just value -> do+      cell <- PQ.getvalue value (PQ.toRow (0 :: Int)) (PQ.toColumn (0 :: Int))+      maybe (fail "count query returned NULL") (pure . read . ByteString.unpack) cell++fixtureName :: Text -> FixtureName+fixtureName = either (error . Text.unpack) id . mkFixtureName++positive :: Int -> Positive+positive = either (error . Text.unpack) id . mkPositive "bench limit"++orFail :: (Show e) => Either e a -> IO a+orFail = either (fail . show) pure
+ examples/tasty-adapter/README.md view
@@ -0,0 +1,5 @@+# Small library adapter++[`TastyAdapter.hs`](TastyAdapter.hs) is compiled in the PostgreSQL test component. Run it with `nix develop -c just test-postgres`. Its `withPreparedManager` function calls `ensureBaseline` once, then keeps one manager and sealed baseline in scope for repeated isolated lease callbacks. `databaseCase` and `runDatabaseCase` show how a test callback returns `Either String ()`: `PreserveFailures` keeps a failed case's clone for inspection, while a successful case releases it. The adapter never changes a callback's value to hide a cleanup error.++For two named databases in one callback, the integration test in `test/Main.hs` calls `withDatabases` with two `DatabaseRequest` values and confirms distinct endpoints. It then gives the second request a bad scenario and verifies that the first clone is unwound. These examples use public Hinagata API types and run on the disposable PostgreSQL cluster supplied by the test script.
+ examples/tasty-adapter/TastyAdapter.hs view
@@ -0,0 +1,37 @@+-- | A small Tasty adapter kept in the test component so the library has no+-- dependency on a particular test framework. Build with+-- @nix develop -c cabal build hinagata-postgres:hinagata-postgres-test@.+module TastyAdapter (withPreparedManager, databaseCase, runDatabaseCase) where++import Data.Either (isLeft)+import Hinagata.Fixture.Bundle (FixturePlan)+import Hinagata.Postgres.Baseline (BaselineError, BaselineRef, BaselineSpec (..), ensureBaseline)+import Hinagata.Postgres.Lease (LeaseDisposition (..), LeaseInfo, LeaseOutcome (..), RetentionPolicy (..))+import Hinagata.Postgres.Manager (Manager, withManagedDatabaseClassified, withManager)+import Test.Tasty (TestTree)+import Test.Tasty.HUnit (assertFailure, testCase)++-- | Prepare once per test group, then reuse the sealed baseline for as many+-- isolated lease callbacks as the caller runs inside this manager scope.+withPreparedManager :: BaselineSpec -> (Manager -> BaselineRef -> IO a) -> IO (Either BaselineError a)+withPreparedManager spec run = do+  prepared <- ensureBaseline spec+  case prepared of+    Left problem -> pure (Left problem)+    Right (baseline, _) -> Right <$> withManager (configuration spec) (\manager -> run manager baseline)++-- | Construct a Tasty case around a prepared suite-scoped manager and frozen+-- fixture plans. A failed assertion returned as 'Left' retains its clone for+-- inspection; a successful case releases it when the callback closes its+-- application connections.+databaseCase :: String -> Manager -> BaselineRef -> FixturePlan -> (LeaseInfo -> IO (Either String ())) -> TestTree+databaseCase name manager baseline scenario check = testCase name (runDatabaseCase manager baseline scenario check)++runDatabaseCase :: Manager -> BaselineRef -> FixturePlan -> (LeaseInfo -> IO (Either String ())) -> IO ()+runDatabaseCase manager baseline scenario check = do+  result <- withManagedDatabaseClassified manager baseline scenario PreserveFailures isLeft check+  case result of+    Left problem -> assertFailure ("database acquisition failed: " ++ show problem)+    Right LeaseOutcome {callbackValue = Left message, leaseInfo} -> assertFailure (message ++ "; retained lease: " ++ show leaseInfo)+    Right LeaseOutcome {callbackValue = Right (), leaseDisposition = LeaseReleased, cleanupDiagnostic = Nothing} -> pure ()+    Right outcome -> assertFailure ("database cleanup failed: " ++ show outcome)
+ hinagata-postgres.cabal view
@@ -0,0 +1,142 @@+cabal-version: 3.12+name: hinagata-postgres+version: 0.1.0.0+synopsis: Transactional PostgreSQL execution for Hinagata fixture plans+description: Private native libpq sessions for atomic SQL and CSV fixtures.+category: Testing+license: BSD-3-Clause+license-file: LICENSE+author: Nadeem Bitar+maintainer: nadeem@gmail.com+homepage: https://github.com/shinzui/hinagata+bug-reports: https://github.com/shinzui/hinagata/issues+extra-source-files: examples/tasty-adapter/README.md+extra-doc-files: CHANGELOG.md+data-files:+  sql/catalog-v1.sql+  sql/catalog-v2.sql+  sql/upgrade-v1-v2.sql+  test/stalled-postgres-peer.py++build-type: Simple++common common+  default-language: GHC2024+  default-extensions:+    DeriveAnyClass+    DuplicateRecordFields+    OverloadedLabels+    OverloadedStrings++  ghc-options: -Wall++library+  import: common+  hs-source-dirs: src+  exposed-modules:+    Hinagata.Postgres.Baseline+    Hinagata.Postgres.Cleanup+    Hinagata.Postgres.Error+    Hinagata.Postgres.Lease+    Hinagata.Postgres.Load+    Hinagata.Postgres.Manager+    Hinagata.Postgres.Ownership+    Hinagata.Postgres.Session++  other-modules:+    Hinagata.Postgres.Internal.Access+    Hinagata.Postgres.Internal.BaselineRef+    Hinagata.Postgres.Internal.Copy+    Hinagata.Postgres.Internal.Libpq+    Hinagata.Postgres.Internal.Load+    Hinagata.Postgres.Internal.Sql+    Paths_hinagata_postgres++  autogen-modules: Paths_hinagata_postgres+  build-depends:+    async >=2.2.6 && <2.3,+    base >=4.21 && <5,+    bytestring >=0.12 && <0.14,+    cryptohash-sha256 >=0.11.102.1 && <0.12,+    directory >=1.3 && <1.5,+    filepath >=1.5 && <1.7,+    hinagata-core ^>=0.1.0.0,+    postgresql-libpq >=0.11.0.0 && <0.12,+    text >=2.1 && <2.3,+    time >=1.14 && <1.16,+    unix >=2.8 && <2.10,++test-suite hinagata-postgres-test+  import: common+  type: exitcode-stdio-1.0+  hs-source-dirs:+    test+    examples/tasty-adapter++  main-is: Main.hs+  other-modules:+    Paths_hinagata_postgres+    TastyAdapter++  autogen-modules: Paths_hinagata_postgres+  ghc-options:+    -O0+    -threaded+    -rtsopts+    -with-rtsopts=-I0++  build-depends:+    base >=4.21 && <5,+    bytestring >=0.12 && <0.14,+    directory >=1.3 && <1.5,+    filepath >=1.5 && <1.7,+    hinagata-core ^>=0.1.0.0,+    hinagata-postgres ^>=0.1.0.0,+    postgresql-libpq >=0.11.0.0 && <0.12,+    process >=1.6.26.1 && <1.7,+    tasty >=1.5.4 && <1.6,+    tasty-hunit >=0.10.2 && <0.11,+    temporary >=1.3 && <1.4,+    text >=2.1 && <2.3,+    unix >=2.8 && <2.10,++benchmark hinagata-postgres-direct-load+  import: common+  type: exitcode-stdio-1.0+  hs-source-dirs: bench+  main-is: Main.hs+  ghc-options: -rtsopts+  build-depends:+    base >=4.21 && <5,+    bytestring >=0.12 && <0.14,+    directory >=1.3 && <1.5,+    filepath >=1.5 && <1.7,+    hinagata-core ^>=0.1.0.0,+    hinagata-postgres ^>=0.1.0.0,+    postgresql-libpq >=0.11.0.0 && <0.12,+    temporary >=1.3 && <1.4,+    text >=2.1 && <2.3,++benchmark hinagata-postgres-concurrent-leases+  import: common+  type: exitcode-stdio-1.0+  hs-source-dirs: bench+  main-is: Concurrency.hs+  ghc-options:+    -threaded+    -rtsopts++  build-depends:+    aeson >=2.2 && <2.3,+    async >=2.2 && <2.3,+    base >=4.21 && <5,+    bytestring >=0.12 && <0.14,+    hinagata-core ^>=0.1.0.0,+    hinagata-postgres ^>=0.1.0.0,+    postgresql-libpq >=0.11.0.0 && <0.12,+    text >=2.1 && <2.3,++source-repository head+  type: git+  location: https://github.com/shinzui/hinagata.git+  subdir: hinagata-postgres
+ sql/catalog-v1.sql view
@@ -0,0 +1,49 @@+-- Transactional schema bootstrap. The caller substitutes one quoted,+-- validated SQL identifier for %SCHEMA% before dispatch.+CREATE SCHEMA %SCHEMA%;++CREATE TABLE %SCHEMA%.meta (+  singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),+  format_version integer NOT NULL CHECK (format_version = 1),+  cluster_uuid uuid NOT NULL+);++INSERT INTO %SCHEMA%.meta (singleton, format_version, cluster_uuid)+VALUES (true, 1, gen_random_uuid());++CREATE TABLE %SCHEMA%.generations (+  id text PRIMARY KEY,+  project_id text NOT NULL,+  fingerprint text NOT NULL,+  fingerprint_manifest jsonb NOT NULL,+  database_name name NOT NULL UNIQUE,+  database_oid oid,+  ownership_token uuid NOT NULL,+  state text NOT NULL CHECK (state IN ('Building', 'Ready', 'Failed', 'Retiring')),+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()+);++CREATE INDEX generations_lookup ON %SCHEMA%.generations (project_id, fingerprint, state);++CREATE TABLE %SCHEMA%.allocations (+  id text PRIMARY KEY,+  project_id text NOT NULL,+  generation_id text NOT NULL REFERENCES %SCHEMA%.generations(id),+  run_id text,+  database_name name NOT NULL UNIQUE,+  database_oid oid,+  ownership_token uuid NOT NULL,+  state text NOT NULL CHECK (state IN ('Allocating', 'Loading', 'Active', 'Detached', 'Preserved', 'Releasing', 'Released', 'CleanupFailed')),+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()+);++CREATE TABLE %SCHEMA%.leases (+  id text PRIMARY KEY,+  allocation_id text NOT NULL UNIQUE REFERENCES %SCHEMA%.allocations(id),+  run_id text NOT NULL,+  state text NOT NULL CHECK (state IN ('Loading', 'Active', 'Detached', 'Preserved', 'Releasing', 'Released', 'CleanupFailed')),+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()+);
+ sql/catalog-v2.sql view
@@ -0,0 +1,51 @@+-- Transactional schema bootstrap. The caller substitutes one quoted,+-- validated SQL identifier for %SCHEMA% before dispatch.+CREATE SCHEMA %SCHEMA%;++CREATE TABLE %SCHEMA%.meta (+  singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),+  format_version integer NOT NULL CHECK (format_version = 2),+  cluster_uuid uuid NOT NULL+);++INSERT INTO %SCHEMA%.meta (singleton, format_version, cluster_uuid)+VALUES (true, 2, gen_random_uuid());++CREATE TABLE %SCHEMA%.generations (+  id text PRIMARY KEY,+  project_id text NOT NULL,+  fingerprint text NOT NULL,+  fingerprint_manifest jsonb NOT NULL,+  database_name name NOT NULL UNIQUE,+  database_oid oid,+  ownership_token uuid NOT NULL,+  state text NOT NULL CHECK (state IN ('Building', 'Ready', 'Failed', 'Retiring')),+  last_error text,+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()+);++CREATE INDEX generations_lookup ON %SCHEMA%.generations (project_id, fingerprint, state);++CREATE TABLE %SCHEMA%.allocations (+  id text PRIMARY KEY,+  project_id text NOT NULL,+  generation_id text NOT NULL REFERENCES %SCHEMA%.generations(id),+  run_id text,+  database_name name NOT NULL UNIQUE,+  database_oid oid,+  ownership_token uuid NOT NULL,+  state text NOT NULL CHECK (state IN ('Allocating', 'Loading', 'Active', 'Detached', 'Preserved', 'Releasing', 'Released', 'CleanupFailed')),+  last_error text,+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()+);++CREATE TABLE %SCHEMA%.leases (+  id text PRIMARY KEY,+  allocation_id text NOT NULL UNIQUE REFERENCES %SCHEMA%.allocations(id),+  run_id text NOT NULL,+  state text NOT NULL CHECK (state IN ('Loading', 'Active', 'Detached', 'Preserved', 'Releasing', 'Released', 'CleanupFailed')),+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()+);
+ sql/upgrade-v1-v2.sql view
@@ -0,0 +1,7 @@+-- Run only inside the owned catalog bootstrap transaction after validating v1.+-- The caller substitutes one quoted, validated SQL identifier for %SCHEMA%.+ALTER TABLE %SCHEMA%.generations ADD COLUMN last_error text;+ALTER TABLE %SCHEMA%.allocations ADD COLUMN last_error text;+ALTER TABLE %SCHEMA%.meta DROP CONSTRAINT meta_format_version_check;+UPDATE %SCHEMA%.meta SET format_version = 2 WHERE singleton AND format_version = 1;+ALTER TABLE %SCHEMA%.meta ADD CONSTRAINT meta_format_version_check CHECK (format_version = 2);
+ src/Hinagata/Postgres/Baseline.hs view
@@ -0,0 +1,492 @@+-- | Build and reuse immutable PostgreSQL template generations.+module Hinagata.Postgres.Baseline+  ( BaselineSpec (..),+    BaselineRef,+    baselineDatabase,+    PreparationKind (..),+    FingerprintComponent (..),+    fingerprintComponents,+    FingerprintComparison (..),+    PreparationReport (..),+    BaselineError (..),+    ensureBaseline,+    retireBaseline,+  )+where++import Control.Exception (IOException, try)+import Crypto.Hash.SHA256 qualified as SHA256+import Data.Bifunctor (first)+import Data.ByteString qualified as ByteString+import Data.Maybe (listToMaybe)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Database.PostgreSQL.LibPQ qualified as PQ+import GHC.Clock (getMonotonicTimeNSec)+import Hinagata.Config+import Hinagata.Connection+import Hinagata.Fixture.Bundle+import Hinagata.Postgres.Error (FixtureLoadContext, LoadError (..), NativeError (..), NativePhase (..), SessionError (..), loadErrorContext)+import Hinagata.Postgres.Internal.Access qualified as Access+import Hinagata.Postgres.Internal.BaselineRef+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Postgres.Load (loadInto)+import Hinagata.Postgres.Ownership+import Hinagata.Prelude+import Hinagata.Types+import Numeric (showHex)+import System.Timeout (timeout)+import Text.Read (readMaybe)++-- | Hooks must close their own database connections before returning.+-- Missing migration or verification revisions force a new generation on every+-- call. Clone preparation runs on each fresh clone before scenario loading.+data BaselineSpec = BaselineSpec+  { configuration :: !HinagataConfig,+    basePlan :: !FixturePlan,+    migrationRevision :: !(Maybe Text),+    verificationRevision :: !(Maybe Text),+    requiredExtensions :: ![Text],+    requiredLocale :: !(Maybe Text),+    migrationHook :: !(ConnectionTarget -> IO (Either Text ())),+    verificationHook :: !(ConnectionTarget -> IO (Either Text ())),+    clonePreparationHook :: !(Maybe (ConnectionTarget -> IO (Either Text ()))),+    compareAgainst :: !(Maybe BaselineRef)+  }++-- | Name of the sealed template database. Consumers should acquire a clone+-- through the lease API instead of connecting to this database.+baselineDatabase :: BaselineRef -> DatabaseName+baselineDatabase BaselineRef {baselineName} = baselineName++-- | Whether this call published a new generation or reused a ready one.+data PreparationKind = Built | Reused+  deriving stock (Eq, Show)++-- | Non-secret categories in the versioned fingerprint manifest.+data FingerprintComponent+  = FingerprintVersion+  | ServerMajor+  | ProjectIdentity+  | BaseFixtures+  | MigrationRevision+  | VerificationRevision+  | ExtensionRequirements+  | LocaleRequirement+  | AdministrationRole+  | SetupRole+  | ApplicationRole+  | SetupPrivileges+  | ApplicationPrivileges+  | SetupSchemaPrivileges+  | ApplicationSchemaPrivileges+  | SetupRoleSettings+  | ApplicationRoleSettings+  deriving stock (Eq, Show)++-- | Changes against the caller's explicitly selected prior generation.+data FingerprintComparison = FingerprintComparison+  { previousFingerprint :: !Text,+    changedComponents :: ![FingerprintComponent]+  }+  deriving stock (Eq, Show)++-- | Build/reuse report. The duration includes hook and fixture work. No+-- comparison is claimed unless the caller selected a prior generation.+data PreparationReport = PreparationReport+  { kind :: !PreparationKind,+    fingerprint :: !Text,+    elapsedMs :: !Int,+    comparison :: !(Maybe FingerprintComparison)+  }+  deriving stock (Eq, Show)++-- | Non-secret operational failure. An incomplete generation remains recorded+-- for explicit inspection instead of being guessed safe to delete.+data BaselineError = BaselineError+  { cause :: !Text,+    sqlState :: !(Maybe Text),+    fixtureLoadContext :: !(Maybe FixtureLoadContext)+  }+  deriving stock (Eq, Show)++-- | Verify a frozen base plan, then reuse a positively identified sealed+-- generation or build one from @template0@. A failed build leaves its record+-- inspectable and never replaces an earlier ready generation. The caller owns+-- migration and verification hooks and must close their connections before+-- returning; unexpected hook exceptions propagate after the admin session+-- closes.+ensureBaseline :: BaselineSpec -> IO (Either BaselineError (BaselineRef, PreparationReport))+ensureBaseline spec@BaselineSpec {configuration} = do+  start <- getMonotonicTimeNSec+  verified <- try @IOException (verifyPlan (basePlan spec))+  case verified of+    Left _ -> pure (Left (BaselineError "base fixture bundle could not be verified" Nothing Nothing))+    Right False -> pure (Left (BaselineError "base fixture bundle changed" Nothing Nothing))+    Right True -> startCatalog start+  where+    startCatalog start = case first accessFailure (Access.adminTarget configuration (maintenanceDatabase configuration)) of+      Left failure -> pure (Left failure)+      Right maintenance -> do+        let options = defaultSessionOptions {operationDeadlineMs = positiveValue (setupDeadlineMs configuration)}+        catalog <- ensureCatalog maintenance (maintenanceSchema configuration) options+        case catalog of+          Left CatalogError {cause, sqlState} -> pure (Left (BaselineError cause sqlState Nothing))+          Right identity -> do+            opened <- withSession maintenance options $ \session ->+              runExclusive session $ \connection sessionOptions -> do+                outcome <- prepare connection maintenance sessionOptions identity spec+                pure (Keep outcome)+            end <- getMonotonicTimeNSec+            pure $ case opened of+              Left failure -> Left (sessionFailure failure)+              Right (Left failure) -> Left (sessionFailure failure)+              Right (Right (Right (baseline, kind, comparison))) -> Right (baseline, PreparationReport kind (baselineFingerprint baseline) (fromIntegral ((end - start) `div` 1000000)) comparison)+              Right (Right (Left failure)) -> Left failure++-- | Retire a positively identified sealed generation under its exclusive+-- generation lock. Existing clones remain independent; the Retiring catalog+-- record is kept as a tombstone for their allocation references. A retry is+-- idempotent after the owned template has already disappeared.+retireBaseline :: HinagataConfig -> BaselineRef -> IO (Either BaselineError ())+retireBaseline configuration baseline =+  case first accessFailure (Access.adminTarget configuration (maintenanceDatabase configuration)) of+    Left problem -> pure (Left problem)+    Right maintenance -> do+      let options = defaultSessionOptions {operationDeadlineMs = positiveValue (setupDeadlineMs configuration)}+      catalog <- inspectCatalog maintenance (maintenanceSchema configuration) options+      case catalog of+        Left CatalogError {cause, sqlState} -> pure (Left (BaselineError cause sqlState Nothing))+        Right identity+          | identity /= clusterIdentity baseline -> pure (Left (BaselineError "baseline belongs to a different catalog or cluster" Nothing Nothing))+          | otherwise -> do+              opened <- withSession maintenance options $ \session ->+                runExclusive session $ \connection sessionOptions -> do+                  deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+                  locked <- queryParamRows connection deadline Sql "SELECT pg_advisory_lock(1212761905, hashtext($1))" [Just (Encoding.encodeUtf8 ("generation:" <> generationId baseline))] 1+                  result <- case locked of+                    Left problem -> pure (Left (nativeFailure problem))+                    Right _ -> retireLocked connection maintenance sessionOptions identity configuration baseline deadline+                  pure (Keep result)+              pure $ case opened of+                Left problem -> Left (sessionFailure problem)+                Right (Left problem) -> Left (sessionFailure problem)+                Right (Right result) -> result++retireLocked :: PQ.Connection -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> HinagataConfig -> BaselineRef -> Deadline -> IO (Either BaselineError ())+retireLocked connection maintenance options identity configuration baseline deadline = do+  let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+      statement = Encoding.encodeUtf8 ("SELECT database_name::text, database_oid::text, ownership_token::text, state FROM " <> table <> " WHERE id = $1 AND project_id = $2")+      args = map (Just . Encoding.encodeUtf8) [generationId baseline, projectIdText (project configuration)]+  rows <- queryParamRows connection deadline Sql statement args 1+  case rows of+    Left problem -> pure (Left (nativeFailure problem))+    Right [[Just database, Just oidBytes, Just tokenBytes, Just state]]+      | database == Encoding.encodeUtf8 (databaseNameText (baselineName baseline))+          && oidBytes == Encoding.encodeUtf8 (Text.pack (show (oid (ownership baseline))))+          && tokenBytes == Encoding.encodeUtf8 (token (ownership baseline))+          && state `elem` ["Ready", "Retiring"] -> do+          evidence <- verifyOwnedDatabase maintenance options identity (ownership baseline)+          case evidence of+            Left CatalogError {cause, sqlState} -> pure (Left (BaselineError cause sqlState Nothing))+            Right OwnershipMismatch -> pure (Left (BaselineError "baseline ownership changed; retirement refused" Nothing Nothing))+            Right OwnershipMissing+              | state == "Retiring" -> pure (Right ())+              | otherwise -> pure (Left (BaselineError "ready baseline is missing; retirement refused" Nothing Nothing))+            Right OwnershipMatches -> do+              marked <- if state == "Retiring" then pure (Right []) else queryParamRows connection deadline Sql (Encoding.encodeUtf8 ("UPDATE " <> table <> " SET state = 'Retiring', updated_at = clock_timestamp() WHERE id = $1 AND state = 'Ready' RETURNING id")) [Just (Encoding.encodeUtf8 (generationId baseline))] 1+              case marked of+                Left problem -> pure (Left (nativeFailure problem))+                Right [[Just _]] | state == "Ready" -> dropTemplate connection options baseline+                Right [] | state == "Retiring" -> dropTemplate connection options baseline+                Right _ -> pure (Left (BaselineError "baseline state changed during retirement" Nothing Nothing))+    Right _ -> pure (Left (BaselineError "baseline generation is missing or has changed identity" Nothing Nothing))++dropTemplate :: PQ.Connection -> SessionOptions -> BaselineRef -> IO (Either BaselineError ())+dropTemplate connection options baseline = do+  deadline <- deadlineAfter (cleanupDeadlineMs options)+  outcome <- query connection deadline Sql (Encoding.encodeUtf8 ("DROP DATABASE " <> Access.quoteDatabase (baselineName baseline) <> " WITH (FORCE)"))+  pure (first nativeFailure outcome)++prepare :: PQ.Connection -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> BaselineSpec -> IO (Either BaselineError (BaselineRef, PreparationKind, Maybe FingerprintComparison))+prepare connection maintenance options identity spec@BaselineSpec {configuration, basePlan, migrationRevision, verificationRevision} = do+  server <- PQ.serverVersion connection+  let manifest = fingerprintManifest server spec+      digest = hex (SHA256.hash (Encoding.encodeUtf8 manifest))+      projectText = projectIdText (project configuration)+      lockName = projectText <> ":" <> digest+      reusable = isJust migrationRevision && isJust verificationRevision+  deadline <- deadlineAfter (operationDeadlineMs options)+  compared <- comparePrevious connection configuration identity (compareAgainst spec) manifest deadline+  case compared of+    Left failure -> pure (Left failure)+    Right comparison -> do+      locked <- queryParamRows connection deadline Sql "SELECT pg_advisory_lock(1212761905, hashtext($1))" [Just (Encoding.encodeUtf8 lockName)] 1+      case locked of+        Left failure -> pure (Left (nativeFailure failure))+        Right _ -> prepareLocked comparison connection maintenance options identity configuration basePlan projectText digest manifest spec reusable deadline++prepareLocked :: Maybe FingerprintComparison -> PQ.Connection -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> HinagataConfig -> FixturePlan -> Text -> Text -> Text -> BaselineSpec -> Bool -> Deadline -> IO (Either BaselineError (BaselineRef, PreparationKind, Maybe FingerprintComparison))+prepareLocked comparison connection maintenance options identity configuration basePlan projectText digest manifest spec reusable deadline = do+  let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+      interruptedSql = Encoding.encodeUtf8 ("UPDATE " <> table <> " SET state = 'Failed', last_error = 'builder session ended before publication', updated_at = clock_timestamp() WHERE project_id = $1 AND fingerprint = $2 AND state = 'Building'")+      interruptedArgs = map (Just . Encoding.encodeUtf8) [projectText, digest]+  interrupted <- queryParamRows connection deadline Sql interruptedSql interruptedArgs 0+  case interrupted of+    Left failure -> pure (Left (nativeFailure failure))+    Right _ -> do+      previous <- if reusable then lookupReady connection configuration basePlan (clonePreparationHook spec) deadline projectText digest else pure (Right Nothing)+      case previous of+        Left failure -> pure (Left failure)+        Right (Just baseline) -> do+          evidence <- verifyOwnedDatabase maintenance options identity (ownership baseline)+          case evidence of+            Right OwnershipMatches -> do+              sealed <- queryParamRows connection deadline Sql "SELECT datallowconn::text FROM pg_database WHERE datname = $1" [Just (Encoding.encodeUtf8 (databaseNameText (baselineDatabase baseline)))] 1+              pure $ case sealed of+                Right [[Just "false"]] -> Right (baseline {clusterIdentity = identity}, Reused, comparison)+                Right _ -> Left (BaselineError "ready generation is not sealed" Nothing Nothing)+                Left failure -> Left (nativeFailure failure)+            Right _ -> pure (Left (BaselineError "ready generation lost positive database ownership" Nothing Nothing))+            Left CatalogError {cause, sqlState} -> pure (Left (BaselineError cause sqlState Nothing))+        Right Nothing -> fmap (fmap (\(baseline, kind) -> (baseline, kind, comparison))) (buildGeneration connection maintenance options identity configuration basePlan projectText digest manifest spec deadline)++comparePrevious :: PQ.Connection -> HinagataConfig -> CatalogIdentity -> Maybe BaselineRef -> Text -> Deadline -> IO (Either BaselineError (Maybe FingerprintComparison))+comparePrevious _ _ _ Nothing _ _ = pure (Right Nothing)+comparePrevious connection configuration identity (Just previous) manifest deadline+  | clusterIdentity previous /= identity = pure (Left (BaselineError "comparison generation belongs to a different catalog or cluster" Nothing Nothing))+  | otherwise = do+      let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+          statement = Encoding.encodeUtf8 ("SELECT fingerprint::text, fingerprint_manifest->>'manifest' FROM " <> table <> " WHERE id = $1 AND project_id = $2")+          args = map (Just . Encoding.encodeUtf8) [generationId previous, projectIdText (project configuration)]+      rows <- queryParamRows connection deadline Sql statement args 1+      pure $ case rows of+        Left failure -> Left (nativeFailure failure)+        Right [[Just recordedFingerprint, Just recordedManifest]]+          | recordedFingerprint == Encoding.encodeUtf8 (baselineFingerprint previous) ->+              let before = Text.lines (Encoding.decodeUtf8 recordedManifest)+                  after = Text.lines manifest+               in if length before /= length fingerprintComponents || length after /= length fingerprintComponents || listToMaybe before `notElem` map Just ["hinagata-fingerprint-v1", "hinagata-fingerprint-v2"]+                    then Left (BaselineError "comparison generation has an unsupported fingerprint manifest" Nothing Nothing)+                    else Right (Just (FingerprintComparison (baselineFingerprint previous) [component | (component, old, new) <- zip3 fingerprintComponents before after, old /= new]))+        Right _ -> Left (BaselineError "comparison generation is missing or has changed fingerprint identity" Nothing Nothing)++-- | Ordered category names shared by version-1 and version-2 fingerprint manifests.+fingerprintComponents :: [FingerprintComponent]+fingerprintComponents =+  [ FingerprintVersion,+    ServerMajor,+    ProjectIdentity,+    BaseFixtures,+    MigrationRevision,+    VerificationRevision,+    ExtensionRequirements,+    LocaleRequirement,+    AdministrationRole,+    SetupRole,+    ApplicationRole,+    SetupPrivileges,+    ApplicationPrivileges,+    SetupSchemaPrivileges,+    ApplicationSchemaPrivileges,+    SetupRoleSettings,+    ApplicationRoleSettings+  ]++lookupReady :: PQ.Connection -> HinagataConfig -> FixturePlan -> Maybe (ConnectionTarget -> IO (Either Text ())) -> Deadline -> Text -> Text -> IO (Either BaselineError (Maybe BaselineRef))+lookupReady connection configuration plan hook deadline project digest = do+  let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+      statement = Encoding.encodeUtf8 ("SELECT id, database_name::text, database_oid::text, ownership_token::text FROM " <> table <> " WHERE project_id = $1 AND fingerprint = $2 AND state = 'Ready' ORDER BY created_at DESC LIMIT 1")+  rows <- queryParamRows connection deadline Sql statement [Just (Encoding.encodeUtf8 project), Just (Encoding.encodeUtf8 digest)] 1+  pure $ case rows of+    Left failure -> Left (nativeFailure failure)+    Right [] -> Right Nothing+    Right [[Just generationId, Just name, Just oid, Just token]] -> do+      database <- maybe (Left (BaselineError "ready generation has an invalid database name" Nothing Nothing)) Right (either (const Nothing) Just (mkDatabaseName (Encoding.decodeUtf8 name)))+      number <- maybe (Left (BaselineError "ready generation has an invalid database OID" Nothing Nothing)) Right (readMaybe (Text.unpack (Encoding.decodeUtf8 oid)))+      let owned = OwnedDatabase database number (Encoding.decodeUtf8 token)+      Right (Just (BaselineRef database (Encoding.decodeUtf8 generationId) digest (CatalogIdentity 2 "") owned plan hook))+    Right _ -> Left (BaselineError "ready generation has incomplete identity" Nothing Nothing)++buildGeneration :: PQ.Connection -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> HinagataConfig -> FixturePlan -> Text -> Text -> Text -> BaselineSpec -> Deadline -> IO (Either BaselineError (BaselineRef, PreparationKind))+buildGeneration connection maintenance options identity configuration plan project digest manifest spec deadline = do+  generated <- queryParamRows connection deadline Sql "SELECT gen_random_uuid()::text, gen_random_uuid()::text" [] 1+  case generated of+    Right [[Just identifier, Just tokenBytes]] -> do+      let generationId = Encoding.decodeUtf8 identifier+          token = Encoding.decodeUtf8 tokenBytes+          databaseText = "hg_" <> Text.take 8 digest <> "_" <> Text.take 20 (Text.filter (/= '-') token)+      case mkDatabaseName databaseText of+        Left _ -> pure (Left (BaselineError "generated database name is invalid" Nothing Nothing))+        Right database -> do+          let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+              insertSql = Encoding.encodeUtf8 ("INSERT INTO " <> table <> " (id, project_id, fingerprint, fingerprint_manifest, database_name, ownership_token, state) VALUES ($1, $2, $3, jsonb_build_object('version', 1, 'manifest', $4::text), $5, $6::uuid, 'Building')")+              args = map (Just . Encoding.encodeUtf8) [generationId, project, digest, manifest, databaseText, token]+          inserted <- queryParamRows connection deadline Sql insertSql args 0+          case inserted of+            Left failure -> pure (Left (nativeFailure failure))+            Right _ -> do+              built <- construct connection maintenance options identity configuration plan generationId database token spec deadline+              case built of+                Left failure@BaselineError {cause = failureCause} -> do+                  let failedSql = Encoding.encodeUtf8 ("UPDATE " <> table <> " SET state = 'Failed', last_error = $2, updated_at = clock_timestamp() WHERE id = $1")+                  _ <- queryParamRows connection deadline Sql failedSql [Just identifier, Just (Encoding.encodeUtf8 (Text.take 512 failureCause))] 0+                  pure (Left failure)+                Right owned -> do+                  let readySql = Encoding.encodeUtf8 ("UPDATE " <> table <> " SET state = 'Ready', database_oid = $2::oid, updated_at = clock_timestamp() WHERE id = $1")+                  published <- queryParamRows connection deadline Sql readySql [Just identifier, Just (Encoding.encodeUtf8 (Text.pack (show (oid owned))))] 0+                  pure $ case published of+                    Left failure -> Left (nativeFailure failure)+                    Right _ -> Right (BaselineRef database generationId digest identity owned plan (clonePreparationHook spec), Built)+    Left failure -> pure (Left (nativeFailure failure))+    Right _ -> pure (Left (BaselineError "UUID generation returned an invalid result" Nothing Nothing))++construct :: PQ.Connection -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> HinagataConfig -> FixturePlan -> Text -> DatabaseName -> Text -> BaselineSpec -> Deadline -> IO (Either BaselineError OwnedDatabase)+construct connection maintenance options identity configuration plan generationId database token BaselineSpec {migrationHook, verificationHook, requiredExtensions, requiredLocale} deadline = do+  let quoted = Access.quoteDatabase database+  created <- query connection deadline Sql (Encoding.encodeUtf8 ("CREATE DATABASE " <> quoted <> " TEMPLATE template0"))+  case created of+    Left NativeError {sqlState = Just "42501"} -> pure (Left (BaselineError "administration role lacks CREATEDB or template access" (Just "42501") Nothing))+    Left failure -> pure (Left (nativeFailure failure))+    Right () -> do+      let marker = "hinagata:v1:" <> clusterUuid identity <> ":" <> token+      commented <- query connection deadline Sql (Encoding.encodeUtf8 ("COMMENT ON DATABASE " <> quoted <> " IS " <> Access.quoteLiteral marker))+      case commented of+        Left failure -> pure (Left (nativeFailure failure))+        Right () -> do+          observed <- queryParamRows connection deadline Sql "SELECT oid::text FROM pg_database WHERE datname = $1" [Just (Encoding.encodeUtf8 (databaseNameText database))] 1+          case observed of+            Right [[Just oidBytes]] -> case readMaybe (Text.unpack (Encoding.decodeUtf8 oidBytes)) of+              Nothing -> pure (Left (BaselineError "created database has an invalid OID" Nothing Nothing))+              Just number -> do+                let owned = OwnedDatabase database number token+                    table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+                    bindSql = Encoding.encodeUtf8 ("UPDATE " <> table <> " SET database_oid = $2::oid, updated_at = clock_timestamp() WHERE id = $1 AND state = 'Building' RETURNING id")+                bound <- queryParamRows connection deadline Sql bindSql [Just (Encoding.encodeUtf8 generationId), Just oidBytes] 1+                case bound of+                  Left failure -> pure (Left (nativeFailure failure))+                  Right [[Just _]] -> do+                    prepared <- first accessFailure <$> Access.prepareAccess connection options configuration database deadline+                    case prepared of+                      Left failure -> pure (Left failure)+                      Right () -> do+                        case first accessFailure (Access.accessToTarget configuration (setup configuration) database) of+                          Left failure -> pure (Left failure)+                          Right setupTarget -> do+                            migrated <- timeout (positiveValue (setupDeadlineMs configuration) * 1000) (migrationHook setupTarget)+                            case migrated of+                              Nothing -> pure (Left (BaselineError "migration hook timed out" Nothing Nothing))+                              Just (Left _) -> pure (Left (BaselineError "migration hook failed" Nothing Nothing))+                              Just (Right ()) -> do+                                loaded <- loadInto setupTarget options plan+                                case loaded of+                                  Left problem@LoadError {sqlState} -> pure (Left (BaselineError "base fixture load failed" sqlState (Just (loadErrorContext problem))))+                                  Right _ -> do+                                    verified <- timeout (positiveValue (setupDeadlineMs configuration) * 1000) (verificationHook setupTarget)+                                    case verified of+                                      Nothing -> pure (Left (BaselineError "verification hook timed out" Nothing Nothing))+                                      Just (Left _) -> pure (Left (BaselineError "verification hook failed" Nothing Nothing))+                                      Just (Right ()) -> do+                                        requirements <- checkRequirements connection options setupTarget database requiredExtensions requiredLocale deadline+                                        case requirements of+                                          Left failure -> pure (Left failure)+                                          Right () -> do+                                            sessions <- queryParamRows connection deadline Sql "SELECT count(*)::text FROM pg_stat_activity WHERE datname = $1" [Just (Encoding.encodeUtf8 (databaseNameText database))] 1+                                            case sessions of+                                              Right [[Just "0"]] -> do+                                                sealed <- query connection deadline Sql (Encoding.encodeUtf8 ("ALTER DATABASE " <> quoted <> " WITH ALLOW_CONNECTIONS false"))+                                                case sealed of+                                                  Left failure -> pure (Left (nativeFailure failure))+                                                  Right () -> do+                                                    remaining <- queryParamRows connection deadline Sql "SELECT count(*)::text FROM pg_stat_activity WHERE datname = $1" [Just (Encoding.encodeUtf8 (databaseNameText database))] 1+                                                    case remaining of+                                                      Right [[Just "0"]] -> do+                                                        evidence <- verifyOwnedDatabase maintenance options identity owned+                                                        pure $ case evidence of+                                                          Right OwnershipMatches -> Right owned+                                                          Right _ -> Left (BaselineError "sealed generation lost positive ownership" Nothing Nothing)+                                                          Left CatalogError {cause, sqlState} -> Left (BaselineError cause sqlState Nothing)+                                                      Right _ -> pure (Left (BaselineError "sealed baseline gained an open session" Nothing Nothing))+                                                      Left failure -> pure (Left (nativeFailure failure))+                                              Right _ -> pure (Left (BaselineError "baseline still has open sessions" Nothing Nothing))+                                              Left failure -> pure (Left (nativeFailure failure))+                  Right _ -> pure (Left (BaselineError "generation identity could not be bound" Nothing Nothing))+            Right _ -> pure (Left (BaselineError "created database is missing from the catalog" Nothing Nothing))+            Left failure -> pure (Left (nativeFailure failure))++checkRequirements :: PQ.Connection -> SessionOptions -> ConnectionTarget -> DatabaseName -> [Text] -> Maybe Text -> Deadline -> IO (Either BaselineError ())+checkRequirements maintenance options setupTarget database extensions locale deadline = do+  localeResult <- case locale of+    Nothing -> pure (Right ())+    Just expected -> do+      observed <- queryParamRows maintenance deadline Sql "SELECT datcollate FROM pg_database WHERE datname = $1" [Just (Encoding.encodeUtf8 (databaseNameText database))] 1+      pure $ case observed of+        Right [[Just actual]] | actual == Encoding.encodeUtf8 expected -> Right ()+        Right _ -> Left (BaselineError "baseline locale differs from the declared requirement" Nothing Nothing)+        Left failure -> Left (nativeFailure failure)+  case localeResult of+    Left failure -> pure (Left failure)+    Right () ->+      if null extensions+        then pure (Right ())+        else do+          opened <- withSession setupTarget options $ \session ->+            runExclusive session $ \connection sessionOptions -> do+              innerDeadline <- deadlineAfter (operationDeadlineMs sessionOptions)+              results <- traverse (checkExtension connection innerDeadline) extensions+              pure (Keep (() <$ sequence results))+          pure $ case opened of+            Left failure -> Left (sessionFailure failure)+            Right (Left failure) -> Left (sessionFailure failure)+            Right (Right result) -> result++checkExtension :: PQ.Connection -> Deadline -> Text -> IO (Either BaselineError ())+checkExtension connection deadline extension = do+  observed <- queryParamRows connection deadline Sql "SELECT extname::text FROM pg_extension WHERE extname = $1" [Just (Encoding.encodeUtf8 extension)] 1+  pure $ case observed of+    Right [[Just _]] -> Right ()+    Right _ -> Left (BaselineError "baseline lacks a declared extension" Nothing Nothing)+    Left failure -> Left (nativeFailure failure)++fingerprintManifest :: Int -> BaselineSpec -> Text+fingerprintManifest server BaselineSpec {configuration, basePlan, migrationRevision, verificationRevision, requiredExtensions, requiredLocale} =+  Text.intercalate+    "\n"+    [ "hinagata-fingerprint-v2",+      render (server `div` 10000),+      render (projectIdText (project configuration)),+      render (planDigest basePlan),+      render migrationRevision,+      render verificationRevision,+      render requiredExtensions,+      render requiredLocale,+      render (user (administration configuration)),+      render (user (setup configuration)),+      render (user (application configuration)),+      render (setupGrants configuration),+      render (applicationGrants configuration),+      render (setupSchemaGrants configuration),+      render (applicationSchemaGrants configuration),+      render (nonSecretSettings (setupSettings configuration)),+      render (nonSecretSettings (applicationSettings configuration))+    ]+  where+    render :: (Show a) => a -> Text+    render = Text.pack . show++    nonSecretSettings :: [RoleSetting] -> [(Text, Text)]+    nonSecretSettings = map (\RoleSetting {name, value} -> (sqlIdentifierText name, hex (SHA256.hash (Encoding.encodeUtf8 value))))++hex :: ByteString.ByteString -> Text+hex = Text.pack . concatMap digits . ByteString.unpack+  where+    digits byte = case showHex byte "" of [digit] -> ['0', digit]; value -> value++nativeFailure :: NativeError -> BaselineError+nativeFailure NativeError {reason, sqlState} = BaselineError reason sqlState Nothing++accessFailure :: Access.AccessError -> BaselineError+accessFailure Access.AccessError {cause, sqlState} = BaselineError cause sqlState Nothing++sessionFailure :: SessionError -> BaselineError+sessionFailure failure = BaselineError (Text.pack (show failure)) Nothing Nothing
+ src/Hinagata/Postgres/Cleanup.hs view
@@ -0,0 +1,426 @@+-- | Preview and explicitly apply ownership-checked clone recovery.+module Hinagata.Postgres.Cleanup+  ( CleanupDisposition (..),+    CleanupCandidate (..),+    CleanupSelection (..),+    CleanupResult (..),+    CleanupError (..),+    planCleanup,+    inspectLease,+    applyCleanup,+    releaseLease,+    preserveLease,+  )+where++import Data.Bifunctor (first)+import Data.ByteString (ByteString)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Database.PostgreSQL.LibPQ qualified as PQ+import Hinagata.Config+import Hinagata.Connection (ConnectionTarget)+import Hinagata.Postgres.Error (NativeError (..), NativePhase (..), SessionError)+import Hinagata.Postgres.Internal.Access qualified as Access+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Postgres.Ownership qualified as Own+import Hinagata.Prelude+import Hinagata.Types+import Text.Read (readMaybe)++-- | Read-only classification. An ambiguous identity is never deletion+-- authority, even when a database happens to have the recorded name.+data CleanupDisposition = Live | Orphaned | Retained | Ambiguous | Missing | Foreign+  deriving stock (Eq, Show)++-- | Catalog identity and evidence at preview time. Apply always rechecks.+data CleanupCandidate = CleanupCandidate+  { allocationId :: !Text,+    leaseId :: !(Maybe Text),+    database :: !DatabaseName,+    state :: !Text,+    disposition :: !CleanupDisposition+  }+  deriving stock (Eq, Show)++-- | Ordinary recovery touches only orphaned active work. Retained resources+-- require an explicit selected ID and the broader policy.+data CleanupSelection = OrphansOnly | IncludeRetained+  deriving stock (Eq, Show)++-- | Per-ID apply outcome. A refusal leaves the database and catalog visible.+data CleanupResult = Released !Text | AlreadyReleased !Text | Preserved !Text | Skipped !Text !CleanupDisposition | Refused !Text !Text+  deriving stock (Eq, Show)++-- | Non-secret preview or apply failure.+data CleanupError = CleanupError+  { cause :: !Text,+    sqlState :: !(Maybe Text)+  }+  deriving stock (Eq, Show)++data Snapshot = Snapshot+  { allocation :: !Text,+    generation :: !Text,+    databaseName :: !DatabaseName,+    databaseOid :: !(Maybe Int),+    ownershipToken :: !Text,+    allocationState :: !Text,+    recordedLease :: !(Maybe Text)+  }++-- | Inspect project allocations in stable batches. A transient try-lock may+-- be taken and released to classify liveness; no catalog row or database is+-- changed. Results are advisory and grant no later deletion authority.+planCleanup :: HinagataConfig -> IO (Either CleanupError [CleanupCandidate])+planCleanup configuration = withCatalog configuration $ \maintenance options identity -> do+  opened <- withSession maintenance options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      candidates <- collect connection maintenance options identity configuration deadline "" []+      pure (Keep candidates)+  pure (flatten opened)++-- | Inspect one lease, including an already released record, without changing+-- catalog or database state. A held callback lock is reported as Live.+inspectLease :: HinagataConfig -> Text -> IO (Either CleanupError (Maybe CleanupCandidate))+inspectLease configuration identifier = withCatalog configuration $ \maintenance options identity -> do+  opened <- withSession maintenance options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      snapshot <- lookupLeaseSnapshot connection configuration deadline identifier+      result <- case snapshot of+        Left problem -> pure (Left problem)+        Right Nothing -> pure (Right Nothing)+        Right (Just found) -> fmap Just <$> classify connection maintenance options identity deadline found+      pure (Keep result)+  pure (flatten opened)++-- | Revalidate each explicitly selected allocation under its generation and+-- lease locks. A live lock, protected target, or mismatched identity refuses+-- deletion. Missing matching clones are idempotently marked Released.+applyCleanup :: HinagataConfig -> CleanupSelection -> [Text] -> IO (Either CleanupError [CleanupResult])+applyCleanup configuration selection identifiers = withCatalog configuration $ \maintenance options identity ->+  sequence <$> traverse (applyOne configuration selection maintenance options identity) identifiers++-- | Explicitly release one retained lease by its public lease ID. The ID is+-- resolved in the project catalog, then apply repeats every ownership check.+releaseLease :: HinagataConfig -> Text -> IO (Either CleanupError CleanupResult)+releaseLease configuration identifier = do+  resolved <- resolveLease configuration identifier+  case resolved of+    Left problem -> pure (Left problem)+    Right Nothing -> pure (Right (Refused identifier "lease ID was not found in this project"))+    Right (Just allocationId) -> do+      released <- applyCleanup configuration IncludeRetained [allocationId]+      pure $ case released of+        Left problem -> Left problem+        Right [result] -> Right result+        Right _ -> Left (failure "lease release returned an invalid result")++-- | Explicitly preserve an ended or orphaned lease. Live callbacks, unbound+-- identities, missing databases, and foreign ownership evidence are refused.+preserveLease :: HinagataConfig -> Text -> IO (Either CleanupError CleanupResult)+preserveLease configuration identifier = do+  resolved <- resolveLease configuration identifier+  case resolved of+    Left problem -> pure (Left problem)+    Right Nothing -> pure (Right (Refused identifier "lease ID was not found in this project"))+    Right (Just allocationId) -> withCatalog configuration $ \maintenance options identity -> preserveOne configuration maintenance options identity allocationId++resolveLease :: HinagataConfig -> Text -> IO (Either CleanupError (Maybe Text))+resolveLease configuration identifier = withCatalog configuration $ \maintenance options _ -> do+  opened <- withSession maintenance options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      let schema = quoteSqlIdentifier (maintenanceSchema configuration)+          statement = Encoding.encodeUtf8 ("SELECT a.id FROM " <> schema <> ".\"leases\" l JOIN " <> schema <> ".\"allocations\" a ON a.id = l.allocation_id WHERE l.id = $1 AND a.project_id = $2")+          args = map (Just . Encoding.encodeUtf8) [identifier, projectIdText (project configuration)]+      rows <- queryParamRows connection deadline Sql statement args 1+      pure+        ( Keep+            ( case rows of+                Left problem -> Left (nativeFailure problem)+                Right [[Just allocationId]] -> Right (Just (Encoding.decodeUtf8 allocationId))+                Right [] -> Right Nothing+                Right _ -> Left (failure "lease identity is ambiguous")+            )+        )+  pure (flatten opened)++withCatalog :: HinagataConfig -> (ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> IO (Either CleanupError a)) -> IO (Either CleanupError a)+withCatalog configuration action =+  case first accessFailure (Access.adminTarget configuration (maintenanceDatabase configuration)) of+    Left problem -> pure (Left problem)+    Right maintenance -> do+      let options = defaultSessionOptions {operationDeadlineMs = positiveValue (acquisitionDeadlineMs configuration)}+      catalog <- Own.inspectCatalog maintenance (maintenanceSchema configuration) options+      case catalog of+        Left problem -> pure (Left (catalogFailure problem))+        Right identity -> action maintenance options identity++collect :: PQ.Connection -> ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> HinagataConfig -> Deadline -> Text -> [CleanupCandidate] -> IO (Either CleanupError [CleanupCandidate])+collect connection maintenance options identity configuration deadline cursor accumulated = do+  let table = quoteSqlIdentifier (maintenanceSchema configuration)+      statement = Encoding.encodeUtf8 ("SELECT a.id, a.generation_id, a.database_name::text, a.database_oid::text, a.ownership_token::text, a.state, l.id FROM " <> table <> ".\"allocations\" a LEFT JOIN " <> table <> ".\"leases\" l ON l.allocation_id = a.id WHERE a.project_id = $1 AND a.state <> 'Released' AND a.id > $2 ORDER BY a.id LIMIT 100")+      args = map (Just . Encoding.encodeUtf8) [projectIdText (project configuration), cursor]+  page <- queryParamRows connection deadline Sql statement args 100+  case page of+    Left problem -> pure (Left (nativeFailure problem))+    Right rows -> case traverse decodeSnapshot rows of+      Left problem -> pure (Left problem)+      Right snapshots -> do+        classified <- traverse (classify connection maintenance options identity deadline) snapshots+        case sequence classified of+          Left problem -> pure (Left problem)+          Right candidates ->+            if length rows < 100+              then pure (Right (reverse accumulated ++ candidates))+              else collect connection maintenance options identity configuration deadline (allocation (last snapshots)) (reverse candidates ++ accumulated)++classify :: PQ.Connection -> ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> Deadline -> Snapshot -> IO (Either CleanupError CleanupCandidate)+classify connection maintenance options identity deadline snapshot = do+  disposition <- case databaseOid snapshot of+    Nothing -> pure (Right Ambiguous)+    Just number -> do+      held <- case recordedLease snapshot of+        Nothing -> pure (Right False)+        Just identifier -> isLeaseHeld connection deadline identifier+      case held of+        Left problem -> pure (Left problem)+        Right True -> pure (Right Live)+        Right False -> do+          evidence <- Own.verifyOwnedDatabase maintenance options identity (Own.OwnedDatabase (databaseName snapshot) number (ownershipToken snapshot))+          pure $ case evidence of+            Left problem -> Left (catalogFailure problem)+            Right Own.OwnershipMatches+              | allocationState snapshot `elem` ["Detached", "Preserved"] -> Right Retained+              | otherwise -> Right Orphaned+            Right Own.OwnershipMissing -> Right Missing+            Right Own.OwnershipMismatch -> Right Foreign+  pure $ CleanupCandidate (allocation snapshot) (recordedLease snapshot) (databaseName snapshot) (allocationState snapshot) <$> disposition++isLeaseHeld :: PQ.Connection -> Deadline -> Text -> IO (Either CleanupError Bool)+isLeaseHeld connection deadline identifier = do+  let key = Encoding.encodeUtf8 (leaseLockKey identifier)+  attempted <- queryParamRows connection deadline Sql "SELECT pg_try_advisory_lock(1212761905, hashtext($1))::text" [Just key] 1+  case attempted of+    Left problem -> pure (Left (nativeFailure problem))+    Right [[Just "false"]] -> pure (Right True)+    Right [[Just "true"]] -> do+      unlocked <- queryParamRows connection deadline Sql "SELECT pg_advisory_unlock(1212761905, hashtext($1))::text" [Just key] 1+      pure $ case unlocked of+        Right [[Just "true"]] -> Right False+        Left problem -> Left (nativeFailure problem)+        Right _ -> Left (failure "lease preview lock could not be released")+    Right _ -> pure (Left (failure "lease preview lock returned an invalid result"))++applyOne :: HinagataConfig -> CleanupSelection -> ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> Text -> IO (Either CleanupError CleanupResult)+applyOne configuration selection maintenance options identity identifier = do+  opened <- withSession maintenance options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      initial <- lookupSnapshot connection configuration deadline identifier+      result <- case initial of+        Left problem -> pure (Left problem)+        Right Nothing -> pure (Right (AlreadyReleased identifier))+        Right (Just firstSnapshot) -> do+          locked <- queryParamRows connection deadline Sql "SELECT pg_advisory_lock(1212761905, hashtext($1))" [Just (Encoding.encodeUtf8 (generationLockKey (generation firstSnapshot)))] 1+          case locked of+            Left problem -> pure (Left (nativeFailure problem))+            Right _ -> do+              refreshed <- lookupSnapshot connection configuration deadline identifier+              case refreshed of+                Left problem -> pure (Left problem)+                Right Nothing -> pure (Right (AlreadyReleased identifier))+                Right (Just snapshot)+                  | generation snapshot == generation firstSnapshot -> applyLocked connection maintenance options identity configuration selection deadline snapshot+                  | otherwise -> pure (Right (Refused identifier "allocation generation changed while acquiring its lock"))+      pure (Keep result)+  pure (flatten opened)++preserveOne :: HinagataConfig -> ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> Text -> IO (Either CleanupError CleanupResult)+preserveOne configuration maintenance options identity identifier = do+  opened <- withSession maintenance options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      initial <- lookupSnapshot connection configuration deadline identifier+      result <- case initial of+        Left problem -> pure (Left problem)+        Right Nothing -> pure (Right (Refused identifier "allocation was not found in this project"))+        Right (Just firstSnapshot) -> do+          locked <- queryParamRows connection deadline Sql "SELECT pg_advisory_lock(1212761905, hashtext($1))" [Just (Encoding.encodeUtf8 (generationLockKey (generation firstSnapshot)))] 1+          case locked of+            Left problem -> pure (Left (nativeFailure problem))+            Right _ -> do+              refreshed <- lookupSnapshot connection configuration deadline identifier+              case refreshed of+                Left problem -> pure (Left problem)+                Right Nothing -> pure (Right (Refused identifier "allocation disappeared while acquiring its lock"))+                Right (Just snapshot)+                  | generation snapshot == generation firstSnapshot -> preserveLocked connection maintenance options identity configuration deadline snapshot+                  | otherwise -> pure (Right (Refused identifier "allocation generation changed while acquiring its lock"))+      pure (Keep result)+  pure (flatten opened)++preserveLocked :: PQ.Connection -> ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> HinagataConfig -> Deadline -> Snapshot -> IO (Either CleanupError CleanupResult)+preserveLocked connection maintenance options identity configuration deadline snapshot+  | allocationState snapshot == "Released" = pure (Right (AlreadyReleased (allocation snapshot)))+  | databaseName snapshot == maintenanceDatabase configuration || databaseNameText (databaseName snapshot) `elem` ["template0", "template1"] = pure (Right (Refused (allocation snapshot) "protected database target"))+  | otherwise = do+      template <- generationDatabase connection configuration deadline (generation snapshot)+      case template of+        Left problem -> pure (Left problem)+        Right Nothing -> pure (Right (Refused (allocation snapshot) "generation record is missing"))+        Right (Just baselineName) | baselineName == databaseName snapshot -> pure (Right (Refused (allocation snapshot) "allocation names its baseline template"))+        Right (Just _) -> do+          free <- case recordedLease snapshot of+            Nothing -> pure (Right False)+            Just lease -> tryHoldLease connection deadline lease+          case free of+            Left problem -> pure (Left problem)+            Right False -> pure (Right (Skipped (allocation snapshot) Live))+            Right True -> case databaseOid snapshot of+              Nothing -> pure (Right (Refused (allocation snapshot) "allocation has no bound database OID"))+              Just number -> do+                evidence <- Own.verifyOwnedDatabase maintenance options identity (Own.OwnedDatabase (databaseName snapshot) number (ownershipToken snapshot))+                case evidence of+                  Left problem -> pure (Left (catalogFailure problem))+                  Right Own.OwnershipMismatch -> pure (Right (Refused (allocation snapshot) "database ownership evidence differs"))+                  Right Own.OwnershipMissing -> pure (Right (Refused (allocation snapshot) "database is missing"))+                  Right Own.OwnershipMatches -> do+                    updated <- if allocationState snapshot == "Preserved" then pure (Right ()) else changeState connection configuration deadline snapshot "Preserved"+                    pure (Preserved (allocation snapshot) <$ updated)++applyLocked :: PQ.Connection -> ConnectionTarget -> SessionOptions -> Own.CatalogIdentity -> HinagataConfig -> CleanupSelection -> Deadline -> Snapshot -> IO (Either CleanupError CleanupResult)+applyLocked connection maintenance options identity configuration selection deadline snapshot+  | allocationState snapshot == "Released" = pure (Right (AlreadyReleased (allocation snapshot)))+  | databaseName snapshot == maintenanceDatabase configuration || databaseNameText (databaseName snapshot) `elem` ["template0", "template1"] = pure (Right (Refused (allocation snapshot) "protected database target"))+  | allocationState snapshot `elem` ["Detached", "Preserved"] && selection == OrphansOnly = pure (Right (Skipped (allocation snapshot) Retained))+  | otherwise = do+      template <- generationDatabase connection configuration deadline (generation snapshot)+      case template of+        Left problem -> pure (Left problem)+        Right Nothing -> pure (Right (Refused (allocation snapshot) "generation record is missing"))+        Right (Just baselineName) | baselineName == databaseName snapshot -> pure (Right (Refused (allocation snapshot) "allocation names its baseline template"))+        Right (Just _) -> do+          free <- case recordedLease snapshot of+            Nothing -> pure (Right True)+            Just lease -> tryHoldLease connection deadline lease+          case free of+            Left problem -> pure (Left problem)+            Right False -> pure (Right (Skipped (allocation snapshot) Live))+            Right True -> case databaseOid snapshot of+              Nothing -> pure (Right (Refused (allocation snapshot) "allocation has no bound database OID"))+              Just number -> do+                evidence <- Own.verifyOwnedDatabase maintenance options identity (Own.OwnedDatabase (databaseName snapshot) number (ownershipToken snapshot))+                case evidence of+                  Left problem -> pure (Left (catalogFailure problem))+                  Right Own.OwnershipMismatch -> pure (Right (Refused (allocation snapshot) "database ownership evidence differs"))+                  Right Own.OwnershipMissing -> finishReleased connection configuration deadline snapshot+                  Right Own.OwnershipMatches -> do+                    releasing <- changeState connection configuration deadline snapshot "Releasing"+                    case releasing of+                      Left problem -> pure (Left problem)+                      Right () -> do+                        dropped <- query connection deadline Sql (Encoding.encodeUtf8 ("DROP DATABASE " <> Access.quoteDatabase (databaseName snapshot) <> " WITH (FORCE)"))+                        case dropped of+                          Left problem -> do+                            _ <- changeState connection configuration deadline snapshot "CleanupFailed"+                            let statement = Encoding.encodeUtf8 ("UPDATE " <> quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"allocations\" SET last_error = $2 WHERE id = $1")+                            _ <- queryParamRows connection deadline Sql statement [Just (Encoding.encodeUtf8 (allocation snapshot)), Just (Encoding.encodeUtf8 (Text.take 512 (reason problem)))] 0+                            pure (Left (nativeFailure problem))+                          Right () -> finishReleased connection configuration deadline snapshot++finishReleased :: PQ.Connection -> HinagataConfig -> Deadline -> Snapshot -> IO (Either CleanupError CleanupResult)+finishReleased connection configuration deadline snapshot = do+  updated <- changeState connection configuration deadline snapshot "Released"+  pure (Released (allocation snapshot) <$ updated)++tryHoldLease :: PQ.Connection -> Deadline -> Text -> IO (Either CleanupError Bool)+tryHoldLease connection deadline identifier = do+  attempted <- queryParamRows connection deadline Sql "SELECT pg_try_advisory_lock(1212761905, hashtext($1))::text" [Just (Encoding.encodeUtf8 (leaseLockKey identifier))] 1+  pure $ case attempted of+    Left problem -> Left (nativeFailure problem)+    Right [[Just "true"]] -> Right True+    Right [[Just "false"]] -> Right False+    Right _ -> Left (failure "lease lock returned an invalid result")++generationDatabase :: PQ.Connection -> HinagataConfig -> Deadline -> Text -> IO (Either CleanupError (Maybe DatabaseName))+generationDatabase connection configuration deadline identifier = do+  let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+      statement = Encoding.encodeUtf8 ("SELECT database_name::text FROM " <> table <> " WHERE id = $1")+  rows <- queryParamRows connection deadline Sql statement [Just (Encoding.encodeUtf8 identifier)] 1+  pure $ case rows of+    Left problem -> Left (nativeFailure problem)+    Right [] -> Right Nothing+    Right [[Just database]] -> Just <$> first failure (mkDatabaseName (Encoding.decodeUtf8 database))+    Right _ -> Left (failure "generation database identity is invalid")++changeState :: PQ.Connection -> HinagataConfig -> Deadline -> Snapshot -> Text -> IO (Either CleanupError ())+changeState connection configuration deadline snapshot next = do+  let schema = quoteSqlIdentifier (maintenanceSchema configuration)+      statement = Encoding.encodeUtf8 ("WITH allocation_change AS (UPDATE " <> schema <> ".\"allocations\" SET state = $2, last_error = CASE WHEN $2 = 'Released' THEN NULL ELSE last_error END, updated_at = clock_timestamp() WHERE id = $1 RETURNING id), lease_change AS (UPDATE " <> schema <> ".\"leases\" SET state = $2, updated_at = clock_timestamp() WHERE allocation_id IN (SELECT id FROM allocation_change) RETURNING id) SELECT allocation_change.id, (SELECT count(*)::text FROM lease_change) FROM allocation_change")+      args = map (Just . Encoding.encodeUtf8) [allocation snapshot, next]+  rows <- queryParamRows connection deadline Sql statement args 1+  pure $ case rows of+    Right [[Just _, Just _]] -> Right ()+    Left problem -> Left (nativeFailure problem)+    Right _ -> Left (failure "allocation state record disappeared")++lookupSnapshot :: PQ.Connection -> HinagataConfig -> Deadline -> Text -> IO (Either CleanupError (Maybe Snapshot))+lookupSnapshot connection configuration deadline identifier = do+  let schema = quoteSqlIdentifier (maintenanceSchema configuration)+      statement = Encoding.encodeUtf8 ("SELECT a.id, a.generation_id, a.database_name::text, a.database_oid::text, a.ownership_token::text, a.state, l.id FROM " <> schema <> ".\"allocations\" a LEFT JOIN " <> schema <> ".\"leases\" l ON l.allocation_id = a.id WHERE a.id = $1 AND a.project_id = $2")+      args = map (Just . Encoding.encodeUtf8) [identifier, projectIdText (project configuration)]+  rows <- queryParamRows connection deadline Sql statement args 1+  pure $ case rows of+    Left problem -> Left (nativeFailure problem)+    Right [] -> Right Nothing+    Right [row] -> Just <$> decodeSnapshot row+    Right _ -> Left (failure "allocation identity is ambiguous")++lookupLeaseSnapshot :: PQ.Connection -> HinagataConfig -> Deadline -> Text -> IO (Either CleanupError (Maybe Snapshot))+lookupLeaseSnapshot connection configuration deadline identifier = do+  let schema = quoteSqlIdentifier (maintenanceSchema configuration)+      statement = Encoding.encodeUtf8 ("SELECT a.id, a.generation_id, a.database_name::text, a.database_oid::text, a.ownership_token::text, a.state, l.id FROM " <> schema <> ".\"allocations\" a JOIN " <> schema <> ".\"leases\" l ON l.allocation_id = a.id WHERE l.id = $1 AND a.project_id = $2")+      args = map (Just . Encoding.encodeUtf8) [identifier, projectIdText (project configuration)]+  rows <- queryParamRows connection deadline Sql statement args 1+  pure $ case rows of+    Left problem -> Left (nativeFailure problem)+    Right [] -> Right Nothing+    Right [row] -> Just <$> decodeSnapshot row+    Right _ -> Left (failure "lease identity is ambiguous")++decodeSnapshot :: [Maybe ByteString] -> Either CleanupError Snapshot+decodeSnapshot [Just identifier, Just generation, Just database, maybeOid, Just token, Just state, maybeLease] = do+  name <- first failure (mkDatabaseName (Encoding.decodeUtf8 database))+  number <- case maybeOid of+    Nothing -> Right Nothing+    Just value -> maybe (Left (failure "allocation has an invalid database OID")) (Right . Just) (readMaybe (Text.unpack (Encoding.decodeUtf8 value)))+  pure Snapshot {allocation = Encoding.decodeUtf8 identifier, generation = Encoding.decodeUtf8 generation, databaseName = name, databaseOid = number, ownershipToken = Encoding.decodeUtf8 token, allocationState = Encoding.decodeUtf8 state, recordedLease = Encoding.decodeUtf8 <$> maybeLease}+decodeSnapshot _ = Left (failure "allocation has incomplete identity")++flatten :: Either SessionError (Either SessionError (Either CleanupError a)) -> Either CleanupError a+flatten = either (Left . sessionFailure) (either (Left . sessionFailure) id)++failure :: Text -> CleanupError+failure cause = CleanupError cause Nothing++nativeFailure :: NativeError -> CleanupError+nativeFailure NativeError {reason, sqlState} = CleanupError reason sqlState++catalogFailure :: Own.CatalogError -> CleanupError+catalogFailure Own.CatalogError {cause, sqlState} = CleanupError cause sqlState++accessFailure :: Access.AccessError -> CleanupError+accessFailure Access.AccessError {cause, sqlState} = CleanupError cause sqlState++sessionFailure :: SessionError -> CleanupError+sessionFailure problem = failure (Text.pack (show problem))++generationLockKey :: Text -> Text+generationLockKey identifier = "generation:" <> identifier++leaseLockKey :: Text -> Text+leaseLockKey identifier = "lease:" <> identifier
+ src/Hinagata/Postgres/Error.hs view
@@ -0,0 +1,69 @@+-- | Secret-safe native session and load failures.+module Hinagata.Postgres.Error+  ( SessionError (..),+    NativePhase (..),+    NativeError (..),+    LoadPhase (..),+    LoadError (..),+    FixtureLoadContext (..),+    loadErrorContext,+  )+where++import Hinagata.Fixture.Types (FixtureName)+import Hinagata.Prelude++-- | Failure to open or exclusively use a Hinagata-owned session.+data SessionError+  = InvalidSessionOptions+  | ConnectionFailed !Text+  | ConnectionTimedOut !Text+  | SessionBusy+  | SessionClosed+  deriving stock (Eq, Show)++-- | Protocol stage that produced a private adapter failure.+data NativePhase = Connect | Begin | SetDeadline | Sql | Copy | Commit | Rollback+  deriving stock (Eq, Ord, Show)++-- | Server-provided messages may contain fixture data and are intentionally absent.+data NativeError = NativeError+  { phase :: !NativePhase,+    sqlState :: !(Maybe Text),+    reason :: !Text+  }+  deriving stock (Eq, Show)++-- | Stage at which an atomic fixture load failed.+data LoadPhase = VerifyBundle | BeginLoad | SetLoadDeadline | ExecuteSql | ExecuteCopy | CommitLoad | RollbackLoad+  deriving stock (Eq, Ord, Show)++-- | Secret-safe load failure. A failed or ambiguous COMMIT retires its session.+-- @cleanupFailure@ records a separate failure to roll back after the first error.+data LoadError = LoadError+  { phase :: !LoadPhase,+    targetIdentity :: !(Maybe Text),+    fixture :: !(Maybe FixtureName),+    stepIndex :: !(Maybe Int),+    cause :: !Text,+    sqlState :: !(Maybe Text),+    cleanupFailure :: !(Maybe Text)+  }+  deriving stock (Eq, Show)++-- | Safe fixture failure context for lifecycle APIs. Endpoint identity and raw+-- server messages are deliberately excluded. Rollback is independent of clone+-- release, which the enclosing lifecycle error reports separately.+data FixtureLoadContext = FixtureLoadContext+  { phase :: !LoadPhase,+    fixture :: !(Maybe FixtureName),+    stepIndex :: !(Maybe Int),+    sqlState :: !(Maybe Text),+    cause :: !Text,+    rollbackFailure :: !(Maybe Text)+  }+  deriving stock (Eq, Show)++loadErrorContext :: LoadError -> FixtureLoadContext+loadErrorContext LoadError {phase, fixture, stepIndex, sqlState, cause, cleanupFailure} =+  FixtureLoadContext {phase, fixture, stepIndex, sqlState, cause, rollbackFailure = cleanupFailure}
+ src/Hinagata/Postgres/Internal/Access.hs view
@@ -0,0 +1,167 @@+-- | Shared database grants and role-specific settings for generations and clones.+module Hinagata.Postgres.Internal.Access+  ( AccessError (..),+    prepareAccess,+    adminTarget,+    accessToTarget,+    quoteDatabase,+    quoteLiteral,+  )+where++import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Database.PostgreSQL.LibPQ qualified as PQ+import Hinagata.Config+import Hinagata.Connection+import Hinagata.Postgres.Error (NativeError (..), NativePhase (..), SessionError (..))+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Prelude+import Hinagata.Types++data AccessError = AccessError+  { cause :: !Text,+    sqlState :: !(Maybe Text)+  }+  deriving stock (Eq, Show)++prepareAccess :: PQ.Connection -> SessionOptions -> HinagataConfig -> DatabaseName -> Deadline -> IO (Either AccessError ())+prepareAccess connection options configuration database deadline = do+  let grants = [(setup configuration, setupGrants configuration), (application configuration, applicationGrants configuration)]+  databaseGrants <- traverse (uncurry (grantDatabase connection deadline database)) grants+  case sequence databaseGrants of+    Left failure -> pure (Left failure)+    Right _ -> do+      admin <- pure (adminTarget configuration database)+      case admin of+        Left failure -> pure (Left failure)+        Right target -> do+          opened <- withSession target options $ \session ->+            runExclusive session $ \inner innerOptions -> do+              innerDeadline <- deadlineAfter (operationDeadlineMs innerOptions)+              outcomes <- traverse (grantSchema inner innerDeadline) ([(setup configuration, item) | item <- setupSchemaGrants configuration] ++ [(application configuration, item) | item <- applicationSchemaGrants configuration])+              pure (Keep (sequence outcomes))+          case opened of+            Left failure -> pure (Left (accessSessionFailure failure))+            Right (Left failure) -> pure (Left (accessSessionFailure failure))+            Right (Right (Left failure)) -> pure (Left failure)+            Right (Right (Right _)) -> do+              setupResult <- applyRoleSettings configuration options database (setup configuration) (setupSettings configuration)+              case setupResult of+                Left failure -> pure (Left failure)+                Right () -> applyRoleSettings configuration options database (application configuration) (applicationSettings configuration)++applyRoleSettings :: HinagataConfig -> SessionOptions -> DatabaseName -> AccessConfig -> [RoleSetting] -> IO (Either AccessError ())+applyRoleSettings _ _ _ _ [] = pure (Right ())+applyRoleSettings configuration options database access settings =+  case (accessToTarget configuration access database, quoteRole (user access)) of+    (Left failure, _) -> pure (Left failure)+    (_, Left failure) -> pure (Left failure)+    (Right target, Right role) -> do+      opened <- withSession target options $ \session ->+        runExclusive session $ \connection sessionOptions -> do+          deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+          let apply [] = pure (Right ())+              apply (entry : rest) = do+                result <- setRoleSetting connection deadline role database entry+                either (pure . Left) (const (apply rest)) result+          begun <- query connection deadline Begin "BEGIN"+          case begun of+            Left failure -> pure (Retire (Left (accessNativeFailure failure)))+            Right () -> do+              result <- apply settings+              case result of+                Right () -> do+                  committed <- query connection deadline Commit "COMMIT"+                  pure $ either (Retire . Left . accessNativeFailure) (const (Keep (Right ()))) committed+                Left failure -> do+                  cleanup <- deadlineAfter (cleanupDeadlineMs sessionOptions)+                  rolled <- query connection cleanup Rollback "ROLLBACK"+                  pure $ case rolled of+                    Right () -> Keep (Left failure)+                    Left _ -> Retire (Left failure)+      pure $ case opened of+        Left failure -> Left (accessSessionFailure failure)+        Right (Left failure) -> Left (accessSessionFailure failure)+        Right (Right result) -> () <$ result++setRoleSetting :: PQ.Connection -> Deadline -> Text -> DatabaseName -> RoleSetting -> IO (Either AccessError ())+setRoleSetting connection deadline role database RoleSetting {name, value}+  | Text.any (== '\0') value = pure (Left (AccessError "role setting contains NUL" Nothing))+  | otherwise = do+      let prefix = "ALTER ROLE " <> role <> " IN DATABASE " <> quoteDatabase database <> " SET " <> quoteSqlIdentifier name+          persist suffix = either (Left . accessNativeFailure) Right <$> query connection deadline Sql (Encoding.encodeUtf8 (prefix <> suffix))+      if Text.toLower (sqlIdentifierText name) `elem` quotedListSettings+        then do+          interpreted <- queryParamRows connection deadline Sql "SELECT pg_catalog.set_config($1, $2, true)" (map (Just . Encoding.encodeUtf8) [sqlIdentifierText name, value]) 1+          case interpreted of+            Left failure -> pure (Left (accessNativeFailure failure))+            Right _ -> persist " FROM CURRENT"+        else persist (" TO " <> quoteSettingLiteral value)++-- Explicit escape-string syntax preserves backslashes independently of the+-- persistence role's standard_conforming_strings default.+quoteSettingLiteral :: Text -> Text+quoteSettingLiteral value = "E'" <> Text.replace "'" "''" (Text.replace "\\" "\\\\" value) <> "'"++-- PostgreSQL's GUC_LIST_QUOTE parameters (also enumerated by pg_dump).+-- A single SQL literal would quote the entire list as one item. Other settings+-- keep scalar assignment so read-only, role, encoding, and timeout declarations+-- cannot change the persistence session before ALTER ROLE executes.+quotedListSettings :: [Text]+quotedListSettings =+  [ "search_path",+    "temp_tablespaces",+    "session_preload_libraries",+    "local_preload_libraries",+    "shared_preload_libraries",+    "unix_socket_directories",+    "oauth_validator_libraries",+    "output_plugin_libraries"+  ]++grantDatabase :: PQ.Connection -> Deadline -> DatabaseName -> AccessConfig -> [DatabaseGrant] -> IO (Either AccessError ())+grantDatabase connection deadline database access grants =+  case quoteRole (user access) of+    Left failure -> pure (Left failure)+    Right role ->+      if null grants+        then pure (Right ())+        else do+          let privileges = Text.intercalate ", " (map grantName grants)+              statement = Encoding.encodeUtf8 ("GRANT " <> privileges <> " ON DATABASE " <> quoteDatabase database <> " TO " <> role)+          either (Left . accessNativeFailure) Right <$> query connection deadline Sql statement++grantSchema :: PQ.Connection -> Deadline -> (AccessConfig, SchemaGrant) -> IO (Either AccessError ())+grantSchema connection deadline (access, SchemaGrant {schema, privilege}) =+  case quoteRole (user access) of+    Left failure -> pure (Left failure)+    Right role -> do+      let name = case privilege of SchemaUsage -> "USAGE"; SchemaCreate -> "CREATE"+          statement = Encoding.encodeUtf8 ("GRANT " <> name <> " ON SCHEMA " <> quoteSqlIdentifier schema <> " TO " <> role)+      either (Left . accessNativeFailure) Right <$> query connection deadline Sql statement++grantName :: DatabaseGrant -> Text+grantName = \case GrantConnect -> "CONNECT"; GrantCreate -> "CREATE"; GrantTemporary -> "TEMPORARY"++adminTarget :: HinagataConfig -> DatabaseName -> Either AccessError ConnectionTarget+adminTarget configuration = accessToTarget configuration (administration configuration)++accessToTarget :: HinagataConfig -> AccessConfig -> DatabaseName -> Either AccessError ConnectionTarget+accessToTarget HinagataConfig {endpoint = EndpointConfig {host, port}} AccessConfig {user, password} database =+  either (Left . (\message -> AccessError message Nothing)) Right (mkConnectionTarget host port user database password)++quoteDatabase :: DatabaseName -> Text+quoteDatabase = (\value -> "\"" <> Text.replace "\"" "\"\"" value <> "\"") . databaseNameText++quoteRole :: Text -> Either AccessError Text+quoteRole role = either (Left . (\message -> AccessError message Nothing)) (Right . quoteSqlIdentifier) (mkSqlIdentifier role)++quoteLiteral :: Text -> Text+quoteLiteral value = "'" <> Text.replace "'" "''" value <> "'"++accessNativeFailure :: NativeError -> AccessError+accessNativeFailure NativeError {reason, sqlState} = AccessError reason sqlState++accessSessionFailure :: SessionError -> AccessError+accessSessionFailure failure = AccessError (Text.pack (show failure)) Nothing
+ src/Hinagata/Postgres/Internal/BaselineRef.hs view
@@ -0,0 +1,31 @@+-- | Shared private generation identity for lifecycle modules.+module Hinagata.Postgres.Internal.BaselineRef+  ( BaselineRef (..),+  )+where++import Hinagata.Connection (ConnectionTarget)+import Hinagata.Fixture.Bundle (FixturePlan)+import Hinagata.Postgres.Ownership (CatalogIdentity, OwnedDatabase)+import Hinagata.Prelude+import Hinagata.Types (DatabaseName)++-- | Private identity of a sealed generation; the public API exposes this+-- type abstractly and only permits observing its database name.+data BaselineRef = BaselineRef+  { baselineName :: !DatabaseName,+    generationId :: !Text,+    baselineFingerprint :: !Text,+    clusterIdentity :: !CatalogIdentity,+    ownership :: !OwnedDatabase,+    baselinePlan :: !FixturePlan,+    cloneHook :: !(Maybe (ConnectionTarget -> IO (Either Text ())))+  }++instance Eq BaselineRef where+  left == right =+    (baselineName left, generationId left, baselineFingerprint left, clusterIdentity left, ownership left, baselinePlan left)+      == (baselineName right, generationId right, baselineFingerprint right, clusterIdentity right, ownership right, baselinePlan right)++instance Show BaselineRef where+  show reference = "BaselineRef " <> show (baselineName reference, generationId reference, baselineFingerprint reference, clusterIdentity reference, ownership reference, baselinePlan reference)
+ src/Hinagata/Postgres/Internal/Copy.hs view
@@ -0,0 +1,156 @@+-- | Bounded COPY FROM STDIN with a single nonblocking libpq protocol owner.+module Hinagata.Postgres.Internal.Copy+  ( copyCsv,+  )+where++import Control.Exception (IOException, try)+import Crypto.Hash.SHA256 qualified as SHA256+import Data.ByteString (ByteString)+import Data.ByteString qualified as ByteString+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Word (Word64)+import Database.PostgreSQL.LibPQ qualified as PQ+import Hinagata.Fixture.Bundle (CapturedStep (..))+import Hinagata.Fixture.Types (CopySpec (..))+import Hinagata.Postgres.Error+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Prelude+import Hinagata.Types (quoteSqlIdentifier)+import Numeric (showHex)+import System.IO (IOMode (ReadMode), withBinaryFile)++copyCsv :: PQ.Connection -> Deadline -> FilePath -> CapturedStep -> IO (Either NativeError Word64)+copyCsv connection deadline path step =+  case step of+    CapturedCopy {copySpec, contentDigest, byteLength} -> do+      begun <- startCopy connection deadline copySpec+      case begun of+        Left failure -> pure (Left failure)+        Right () -> do+          transferred <- try @IOException (withBinaryFile path ReadMode (stream byteLength contentDigest))+          case transferred of+            Left _ -> do+              abortCopy connection deadline+              pure (Left (NativeError Copy Nothing "COPY bundle file could not be read"))+            Right (Left failure) -> do+              abortCopy connection deadline+              pure (Left failure)+            Right (Right count) -> do+              ended <- endCopy connection deadline+              pure (count <$ ended)+    _ -> pure (Left (NativeError Copy Nothing "expected captured CSV step"))+  where+    stream expected digest input = go input 0 SHA256.init+      where+        go handle count context = do+          bytes <- ByteString.hGet handle 65536+          if ByteString.null bytes+            then+              if count == expected && hex (SHA256.finalize context) == digest+                then pure (Right count)+                else pure (Left (NativeError Copy Nothing "COPY bundle bytes changed"))+            else do+              sent <- sendChunk connection deadline bytes+              case sent of+                Left failure -> pure (Left failure)+                Right () -> do+                  let nextCount = count + fromIntegral (ByteString.length bytes)+                      nextContext = SHA256.update context bytes+                  nextContext `seq` go handle nextCount nextContext++startCopy :: PQ.Connection -> Deadline -> CopySpec -> IO (Either NativeError ())+startCopy connection deadline spec = do+  sent <- PQ.sendQuery connection (renderCopy spec)+  if not sent+    then pure (Left (NativeError Copy Nothing "libpq refused COPY dispatch"))+    else do+      flushed <- flushOutput connection deadline+      if not flushed+        then pure (Left (NativeError Copy Nothing "COPY dispatch timed out"))+        else do+          ready <- awaitResult connection deadline+          if not ready+            then pure (Left (NativeError Copy Nothing "COPY start timed out"))+            else do+              result <- PQ.getResult connection+              case result of+                Nothing -> pure (Left (NativeError Copy Nothing "COPY returned no start result"))+                Just value -> do+                  status <- PQ.resultStatus value+                  if status == PQ.CopyIn+                    then pure (Right ())+                    else do+                      state <- PQ.resultErrorField value PQ.DiagSqlstate+                      _ <- drainResults connection deadline Copy+                      pure (Left (NativeError Copy (Encoding.decodeUtf8 <$> state) ("COPY start returned " <> Text.pack (show status))))++sendChunk :: PQ.Connection -> Deadline -> ByteString -> IO (Either NativeError ())+sendChunk connection deadline bytes = do+  outcome <- PQ.putCopyData connection bytes+  case outcome of+    PQ.CopyInOk -> do+      -- Nonblocking libpq queues successful COPY writes until PQflush. Drain+      -- each bounded chunk so client memory cannot grow with fixture size.+      flushed <- flushOutput connection deadline+      pure $ if flushed then Right () else Left (NativeError Copy Nothing "COPY data flush timed out")+    PQ.CopyInError -> pure (Left (NativeError Copy Nothing "COPY data transfer failed"))+    PQ.CopyInWouldBlock -> do+      flushed <- flushOutput connection deadline+      if not flushed+        then pure (Left (NativeError Copy Nothing "COPY data flush timed out"))+        else do+          ready <- awaitWritable connection deadline+          if ready then sendChunk connection deadline bytes else pure (Left (NativeError Copy Nothing "COPY transfer timed out"))++endCopy :: PQ.Connection -> Deadline -> IO (Either NativeError ())+endCopy connection deadline = do+  outcome <- PQ.putCopyEnd connection Nothing+  case outcome of+    PQ.CopyInOk -> do+      flushed <- flushOutput connection deadline+      if flushed then drainResults connection deadline Copy else pure (Left (NativeError Copy Nothing "COPY end flush timed out"))+    PQ.CopyInError -> pure (Left (NativeError Copy Nothing "COPY end failed"))+    PQ.CopyInWouldBlock -> do+      flushed <- flushOutput connection deadline+      if not flushed+        then pure (Left (NativeError Copy Nothing "COPY end flush timed out"))+        else do+          ready <- awaitWritable connection deadline+          if ready then endCopy connection deadline else pure (Left (NativeError Copy Nothing "COPY end timed out"))++abortCopy :: PQ.Connection -> Deadline -> IO ()+abortCopy connection deadline = do+  outcome <- PQ.putCopyEnd connection (Just "Hinagata bundle transfer failed")+  case outcome of+    PQ.CopyInOk -> do+      _ <- flushOutput connection deadline+      _ <- drainResults connection deadline Copy+      pure ()+    PQ.CopyInWouldBlock -> do+      flushed <- flushOutput connection deadline+      when flushed $ do+        ready <- awaitWritable connection deadline+        when ready (abortCopy connection deadline)+    PQ.CopyInError -> pure ()++renderCopy :: CopySpec -> ByteString+renderCopy CopySpec {schema, table, columns, header} =+  Encoding.encodeUtf8 $+    "COPY "+      <> quoteSqlIdentifier schema+      <> "."+      <> quoteSqlIdentifier table+      <> " ("+      <> Text.intercalate ", " (map quoteSqlIdentifier columns)+      <> ") FROM STDIN WITH (FORMAT csv, HEADER "+      <> (if header then "true" else "false")+      <> ")"++hex :: ByteString -> Text+hex = Text.pack . concatMap digits . ByteString.unpack+  where+    digits byte = case showHex byte "" of+      [digit] -> ['0', digit]+      value -> value
+ src/Hinagata/Postgres/Internal/Libpq.hs view
@@ -0,0 +1,279 @@+-- | The sole owner of a libpq connection and its protocol state.+module Hinagata.Postgres.Internal.Libpq+  ( Session,+    SessionOptions (..),+    defaultSessionOptions,+    sessionIdentity,+    withSession,+    runExclusive,+    SessionDisposition (..),+    query,+    queryParamRows,+    drainResults,+    awaitResult,+    flushOutput,+    awaitReadable,+    awaitWritable,+    Deadline,+    deadlineAfter,+  )+where++import Control.Concurrent (forkIOWithUnmask)+import Control.Concurrent.MVar+import Control.Exception (IOException, SomeException, finally, mask, onException, throwIO, try)+import Data.ByteString (ByteString)+import Data.ByteString qualified as ByteString+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Word (Word64)+import Database.PostgreSQL.LibPQ qualified as PQ+import GHC.Clock (getMonotonicTimeNSec)+import GHC.Conc (threadWaitRead, threadWaitWrite)+import Hinagata.Connection+import Hinagata.Postgres.Error+import Hinagata.Prelude+import System.Posix.Types (Fd)+import System.Timeout (timeout)++-- | Positive timeouts in milliseconds. The operation deadline covers the+-- whole transaction; cleanup gets a fresh, separately bounded deadline.+data SessionOptions = SessionOptions+  { connectDeadlineMs :: !Int,+    operationDeadlineMs :: !Int,+    statementDeadlineMs :: !Int,+    lockDeadlineMs :: !Int,+    cleanupDeadlineMs :: !Int+  }+  deriving stock (Eq, Show)++-- | Five-second connect/cleanup deadlines, five-minute operation and+-- statement deadlines, and a 30-second lock deadline.+defaultSessionOptions :: SessionOptions+defaultSessionOptions = SessionOptions 5000 300000 300000 30000 5000++data State = Open | Closed+  deriving stock (Eq, Show)++-- | Opaque owner of a single native connection. Use 'withSession' to bracket it.+data Session = Session+  { connection :: !PQ.Connection,+    gate :: !(MVar State),+    options :: !SessionOptions,+    identity :: !Text+  }++sessionIdentity :: Session -> Text+sessionIdentity Session {identity} = identity++data SessionDisposition a = Keep !a | Retire !a++newtype Deadline = Deadline Word64++deadlineAfter :: Int -> IO Deadline+deadlineAfter milliseconds = do+  now <- getMonotonicTimeNSec+  pure (Deadline (now + fromIntegral (max 1 milliseconds) * 1000000))++-- | Open and bracket a connection to the explicit target. A failed Unix socket+-- connection does not retry through TCP. Callback exceptions propagate after+-- the connection closes. Interrupted acquisition finishes its native handle+-- before returning, without relying on garbage collection.+withSession :: ConnectionTarget -> SessionOptions -> (Session -> IO a) -> IO (Either SessionError a)+withSession target options action = mask $ \restore -> do+  opened <- openSession target options+  case opened of+    Left failure -> pure (Left failure)+    Right session -> Right <$> (restore (action session) `finally` closeSession session)++openSession :: ConnectionTarget -> SessionOptions -> IO (Either SessionError Session)+openSession target options+  | any (<= 0) [connectDeadlineMs options, operationDeadlineMs options, statementDeadlineMs options, lockDeadlineMs options, cleanupDeadlineMs options] = pure (Left InvalidSessionOptions)+  | otherwise = mask $ \restore -> do+      let identity = Text.pack (show target)+          connectionInfo = Encoding.encodeUtf8 (connectionStringText (renderConnectionString target))+      started <- try @IOException (PQ.connectStart connectionInfo)+      case started of+        Left _ -> pure (Left (ConnectionFailed identity))+        Right connection ->+          ( do+              deadline <- deadlineAfter (connectDeadlineMs options)+              outcome <- try @IOException (restore (poll connection deadline))+              case outcome of+                Left _ -> PQ.finish connection >> pure (Left (ConnectionFailed identity))+                Right False -> PQ.finish connection >> pure (Left (ConnectionTimedOut identity))+                Right True -> do+                  nonblocking <- PQ.setnonblocking connection True+                  if not nonblocking+                    then PQ.finish connection >> pure (Left (ConnectionFailed identity))+                    else do+                      gate <- newMVar Open+                      pure (Right Session {connection, gate, options, identity})+          )+            `onException` PQ.finish connection+  where+    poll connection deadline = do+      status <- PQ.connectPoll connection+      case status of+        PQ.PollingOk -> pure True+        PQ.PollingFailed -> pure False+        PQ.PollingReading -> awaitReadable connection deadline >>= \ready -> if ready then poll connection deadline else pure False+        PQ.PollingWriting -> awaitWritable connection deadline >>= \ready -> if ready then poll connection deadline else pure False++closeSession :: Session -> IO ()+closeSession Session {connection, gate} =+  modifyMVar_ gate $ \case+    Open -> PQ.finish connection >> pure Closed+    Closed -> pure Closed++-- | A simultaneous use fails immediately. An asynchronous exception closes+-- the connection, causing PostgreSQL to roll back its open transaction.+runExclusive :: Session -> (PQ.Connection -> SessionOptions -> IO (SessionDisposition a)) -> IO (Either SessionError a)+runExclusive Session {connection, gate, options} action = mask $ \restore -> do+  acquired <- tryTakeMVar gate+  case acquired of+    Nothing -> pure (Left SessionBusy)+    Just Closed -> putMVar gate Closed >> pure (Left SessionClosed)+    Just Open -> do+      result <- try @SomeException (restore (action connection options))+      case result of+        Right (Keep value) -> putMVar gate Open >> pure (Right value)+        Right (Retire value) -> do+          PQ.finish connection `finally` putMVar gate Closed+          pure (Right value)+        Left failure -> do+          requestCancel connection+          PQ.finish connection `finally` putMVar gate Closed+          throwIO failure++-- | Cancellation is best-effort and bounded. Closing the connection remains+-- the authoritative rollback when the protocol was interrupted.+requestCancel :: PQ.Connection -> IO ()+requestCancel connection = do+  handle <- try @SomeException (PQ.getCancel connection)+  case handle of+    Left _ -> pure ()+    Right Nothing -> pure ()+    Right (Just cancelHandle) -> do+      finished <- newEmptyMVar+      _ <- forkIOWithUnmask $ \unmask -> do+        _ <- try @SomeException (unmask (PQ.cancel cancelHandle))+        putMVar finished ()+      void (timeout 1000000 (takeMVar finished))++-- | Send and drain a query without retaining result rows. Result status is+-- checked for every statement, including the final COMMIT result.+query :: PQ.Connection -> Deadline -> NativePhase -> ByteString -> IO (Either NativeError ())+query connection deadline phase statement = do+  sent <- PQ.sendQuery connection statement+  if not sent+    then pure (Left (NativeError phase Nothing "libpq refused query dispatch"))+    else do+      singleRows <- PQ.setSingleRowMode connection+      if not singleRows+        then pure (Left (NativeError phase Nothing "libpq refused single-row result mode"))+        else do+          flushed <- flushOutput connection deadline+          if not flushed+            then pure (Left (NativeError phase Nothing "query output could not be flushed before deadline"))+            else drainResults connection deadline phase++-- | Execute one parameterized catalog query, retaining at most a small,+-- explicit number of copied result cells. Single-row mode keeps libpq from+-- constructing a large result before the bound can be checked.+queryParamRows :: PQ.Connection -> Deadline -> NativePhase -> ByteString -> [Maybe ByteString] -> Int -> IO (Either NativeError [[Maybe ByteString]])+queryParamRows connection deadline phase statement arguments maxRows = do+  sent <- PQ.sendQueryParams connection statement (map (fmap (\value -> (PQ.Oid 0, value, PQ.Text))) arguments) PQ.Text+  if not sent+    then pure (Left (NativeError phase Nothing "libpq refused catalog query dispatch"))+    else do+      singleRows <- PQ.setSingleRowMode connection+      if not singleRows+        then pure (Left (NativeError phase Nothing "libpq refused catalog single-row mode"))+        else do+          flushed <- flushOutput connection deadline+          if not flushed+            then pure (Left (NativeError phase Nothing "catalog query output could not be flushed before deadline"))+            else collect [] Nothing+  where+    collect rows firstError = do+      ready <- awaitResult connection deadline+      if not ready+        then pure (Left (NativeError phase Nothing "catalog query timed out or connection failed"))+        else do+          next <- PQ.getResult connection+          case next of+            Nothing -> pure (maybe (Right (reverse rows)) Left firstError)+            Just result -> do+              status <- PQ.resultStatus result+              state <- PQ.resultErrorField result PQ.DiagSqlstate+              case status of+                PQ.SingleTuple -> do+                  columns <- PQ.nfields result+                  if length rows >= maxRows || columns > 20+                    then collect rows (firstError <|> Just (NativeError phase Nothing "catalog query exceeded its result bound"))+                    else do+                      cells <- traverse (PQ.getvalue' result 0) [0 .. columns - 1]+                      if any (maybe False ((> 16384) . ByteString.length)) cells+                        then collect rows (firstError <|> Just (NativeError phase Nothing "catalog query cell exceeded its size bound"))+                        else collect (cells : rows) firstError+                PQ.TuplesOk -> collect rows firstError+                PQ.CommandOk -> collect rows firstError+                _ -> collect rows (firstError <|> Just (NativeError phase (Encoding.decodeUtf8 <$> state) (Text.pack (show status))))++drainResults :: PQ.Connection -> Deadline -> NativePhase -> IO (Either NativeError ())+drainResults connection deadline phase = collect False Nothing+  where+    collect seenResult firstError = do+      ready <- awaitResult connection deadline+      if not ready+        then pure (Left (NativeError phase Nothing "query timed out or connection failed"))+        else do+          next <- PQ.getResult connection+          case next of+            Nothing -> pure (maybe (if seenResult then Right () else Left (NativeError phase Nothing "query returned no result")) Left firstError)+            Just result -> do+              status <- PQ.resultStatus result+              state <- PQ.resultErrorField result PQ.DiagSqlstate+              let failure =+                    if status `elem` [PQ.CommandOk, PQ.TuplesOk, PQ.SingleTuple]+                      then firstError+                      else firstError <|> Just (NativeError phase (Encoding.decodeUtf8 <$> state) (Text.pack (show status)))+              collect True failure++awaitResult :: PQ.Connection -> Deadline -> IO Bool+awaitResult connection deadline = do+  busy <- PQ.isBusy connection+  if not busy+    then pure True+    else do+      ready <- awaitReadable connection deadline+      if not ready+        then pure False+        else do+          consumed <- PQ.consumeInput connection+          if consumed then awaitResult connection deadline else pure False++flushOutput :: PQ.Connection -> Deadline -> IO Bool+flushOutput connection deadline = do+  status <- PQ.flush connection+  case status of+    PQ.FlushOk -> pure True+    PQ.FlushFailed -> pure False+    PQ.FlushWriting -> awaitWritable connection deadline >>= \ready -> if ready then flushOutput connection deadline else pure False++awaitReadable :: PQ.Connection -> Deadline -> IO Bool+awaitReadable connection deadline = waitFor threadWaitRead connection deadline++awaitWritable :: PQ.Connection -> Deadline -> IO Bool+awaitWritable connection deadline = waitFor threadWaitWrite connection deadline++waitFor :: (Fd -> IO ()) -> PQ.Connection -> Deadline -> IO Bool+waitFor operation connection (Deadline limit) = do+  now <- getMonotonicTimeNSec+  descriptor <- PQ.socket connection+  case descriptor of+    Nothing -> pure False+    Just fd+      | now >= limit -> pure False+      | otherwise -> isJust <$> timeout (fromIntegral (min (fromIntegral (maxBound :: Int)) ((limit - now) `div` 1000))) (operation fd)
+ src/Hinagata/Postgres/Internal/Load.hs view
@@ -0,0 +1,202 @@+-- | Atomic direct loading into a caller-selected, already migrated database.+module Hinagata.Postgres.Internal.Load+  ( LoadReport (..),+    loadPlan,+    loadInto,+    loadComposedRemainder,+  )+where++import Control.Exception (IOException, try)+import Crypto.Hash.SHA256 qualified as SHA256+import Data.ByteString qualified as ByteString+import Data.Text qualified as Text+import Data.Word (Word64)+import Database.PostgreSQL.LibPQ qualified as PQ+import GHC.Clock (getMonotonicTimeNSec)+import Hinagata.Connection (ConnectionTarget)+import Hinagata.Fixture.Bundle+import Hinagata.Fixture.Types (FixtureName)+import Hinagata.Postgres.Error (LoadError (..), LoadPhase (..), NativeError (..), NativePhase (..), SessionError)+import Hinagata.Postgres.Internal.Copy (copyCsv)+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Postgres.Internal.Sql+import Hinagata.Prelude+import Numeric (showHex)+import System.FilePath ((</>))+import System.IO (IOMode (ReadMode), withBinaryFile)++-- | Counts and wall-clock stages for a committed fixture plan. @bytesSent@+-- counts transferred SQL and CSV bytes, not inferred database rows.+data LoadReport = LoadReport+  { fixtureCount :: !Int,+    stepCount :: !Int,+    bytesSent :: !Word64,+    verifyMs :: !Word64,+    setupMs :: !Word64,+    sqlMs :: !Word64,+    copyMs :: !Word64,+    commitMs :: !Word64,+    elapsedMs :: !Word64+  }+  deriving stock (Eq, Show)++-- | Open one session, load a verified plan, and close the session. The target+-- database must already exist and contain the schema expected by the fixture.+-- This operation never deletes or resets it.+loadInto :: ConnectionTarget -> SessionOptions -> FixturePlan -> IO (Either LoadError LoadReport)+loadInto target options plan = do+  outcome <- withSession target options (\session -> loadPlan session plan)+  pure $ case outcome of+    Left failure -> Left ((sessionError failure) {targetIdentity = Just (Text.pack (show target))})+    Right value -> value++-- | Load a plan into a reusable Hinagata session under one transaction.+-- Concurrent calls on that session fail with @SessionBusy@; an uncertain+-- connection is retired instead of returned to the caller for reuse.+loadPlan :: Session -> FixturePlan -> IO (Either LoadError LoadReport)+loadPlan session plan = loadSelected True session plan (map (CapturedFixtureRef (planDirectory plan)) (planFixtures plan))++-- | Load only the exact suffix returned by composition. The lease path has+-- already verified its manifest and selected steps; streamed bytes are checked+-- against captured digests again as they are used. This stays private.+loadComposedRemainder :: Session -> FixturePlan -> ComposedPlan -> IO (Either LoadError LoadReport)+loadComposedRemainder session plan composed = loadSelected False session plan (scenarioRemainder composed)++loadSelected :: Bool -> Session -> FixturePlan -> [CapturedFixtureRef] -> IO (Either LoadError LoadReport)+loadSelected verifyBeforeLoad session plan selected = do+  verifyStart <- getMonotonicTimeNSec+  verified <- try @IOException (if verifyBeforeLoad then verifyPlan plan else pure True)+  case verified of+    Left _ -> pure (Left ((simpleError VerifyBundle "bundle verification failed") {targetIdentity = Just (sessionIdentity session)}))+    Right False -> pure (Left ((simpleError VerifyBundle "bundle changed or is incomplete") {targetIdentity = Just (sessionIdentity session)}))+    Right True -> do+      verifyEnd <- getMonotonicTimeNSec+      outcome <- runExclusive session (\connection options -> loadExclusive connection options selected ((verifyEnd - verifyStart) `div` 1000000))+      pure $ case outcome of+        Left failure -> Left ((sessionError failure) {targetIdentity = Just (sessionIdentity session)})+        Right value -> either (Left . (\failure -> failure {targetIdentity = Just (sessionIdentity session)})) Right value++loadExclusive :: PQ.Connection -> SessionOptions -> [CapturedFixtureRef] -> Word64 -> IO (SessionDisposition (Either LoadError LoadReport))+loadExclusive connection options selected verifyMs = do+  transaction <- PQ.transactionStatus connection+  if transaction /= PQ.TransIdle+    then pure (Keep (Left (simpleError BeginLoad "session is already in a transaction or failed state")))+    else do+      start <- getMonotonicTimeNSec+      deadline <- deadlineAfter (operationDeadlineMs options)+      begun <- query connection deadline Begin "BEGIN"+      case begun of+        Left problem -> pure (Retire (Left (nativeError BeginLoad Nothing Nothing problem)))+        Right () -> do+          limits <- setLocalDeadlines connection options deadline+          case limits of+            Left problem -> rollbackAfter connection options (nativeError SetLoadDeadline Nothing Nothing problem)+            Right () -> do+              setupEnd <- getMonotonicTimeNSec+              loaded <- runFixtures connection deadline selected+              case loaded of+                Left failure -> rollbackAfter connection options failure+                Right (steps, bytes, sqlNanos, copyNanos) -> do+                  commitStart <- getMonotonicTimeNSec+                  committed <- query connection deadline Commit "COMMIT"+                  case committed of+                    Left problem -> pure (Retire (Left (nativeError CommitLoad Nothing Nothing problem)))+                    Right () -> do+                      state <- PQ.transactionStatus connection+                      if state /= PQ.TransIdle+                        then pure (Retire (Left (simpleError CommitLoad "connection did not return to idle after commit")))+                        else do+                          end <- getMonotonicTimeNSec+                          pure+                            ( Keep+                                ( Right+                                    LoadReport+                                      { fixtureCount = length selected,+                                        stepCount = steps,+                                        bytesSent = bytes,+                                        verifyMs,+                                        setupMs = (setupEnd - start) `div` 1000000,+                                        sqlMs = sqlNanos `div` 1000000,+                                        copyMs = copyNanos `div` 1000000,+                                        commitMs = (end - commitStart) `div` 1000000,+                                        elapsedMs = verifyMs + (end - start) `div` 1000000+                                      }+                                )+                            )++rollbackAfter :: PQ.Connection -> SessionOptions -> LoadError -> IO (SessionDisposition (Either LoadError LoadReport))+rollbackAfter connection options failure = do+  deadline <- deadlineAfter (cleanupDeadlineMs options)+  rollback <- query connection deadline Rollback "ROLLBACK"+  case rollback of+    Right () -> do+      state <- PQ.transactionStatus connection+      if state == PQ.TransIdle+        then pure (Keep (Left failure))+        else pure (Retire (Left failure {cleanupFailure = Just "connection did not return to idle after rollback"}))+    Left cleanup -> pure (Retire (Left failure {cleanupFailure = Just (reason cleanup)}))++runFixtures :: PQ.Connection -> Deadline -> [CapturedFixtureRef] -> IO (Either LoadError (Int, Word64, Word64, Word64))+runFixtures connection deadline selected = go 0 0 0 0 selected+  where+    go count bytes sqlNanos copyNanos [] = pure (Right (count, bytes, sqlNanos, copyNanos))+    go count bytes sqlNanos copyNanos (CapturedFixtureRef directory captured : rest) = do+      next <- runSteps directory (name captured) 0 count bytes sqlNanos copyNanos (steps captured)+      case next of+        Left failure -> pure (Left failure)+        Right (newCount, newBytes, newSqlNanos, newCopyNanos) -> go newCount newBytes newSqlNanos newCopyNanos rest++    runSteps _ _ _ count bytes sqlNanos copyNanos [] = pure (Right (count, bytes, sqlNanos, copyNanos))+    runSteps directory fixture stepNumber count bytes sqlNanos copyNanos (step : rest) =+      case step of+        CapturedSql {} -> do+          stepStart <- getMonotonicTimeNSec+          content <- readSqlStep directory step+          case content of+            Left failure -> pure (Left failure {fixture = Just fixture, stepIndex = Just stepNumber})+            Right sql -> do+              result <- query connection deadline Sql sql+              case result of+                Left problem -> pure (Left (nativeError ExecuteSql (Just fixture) (Just stepNumber) problem))+                Right () -> do+                  stepEnd <- getMonotonicTimeNSec+                  runSteps directory fixture (stepNumber + 1) (count + 1) (bytes + fromIntegral (ByteString.length sql)) (sqlNanos + stepEnd - stepStart) copyNanos rest+        CapturedCopy {} -> do+          stepStart <- getMonotonicTimeNSec+          let path = directory </> relativePath step+          copied <- copyCsv connection deadline path step+          case copied of+            Left problem -> pure (Left (nativeError ExecuteCopy (Just fixture) (Just stepNumber) problem))+            Right transferred -> do+              stepEnd <- getMonotonicTimeNSec+              runSteps directory fixture (stepNumber + 1) (count + 1) (bytes + transferred) sqlNanos (copyNanos + stepEnd - stepStart) rest++readSqlStep :: FilePath -> CapturedStep -> IO (Either LoadError ByteString.ByteString)+readSqlStep directory step = do+  let path = directory </> relativePath step+  readResult <- try @IOException $ withBinaryFile path ReadMode $ \handle ->+    ByteString.hGet handle (fromIntegral (byteLength step) + 1)+  pure $ case readResult of+    Left _ -> Left (simpleError VerifyBundle "SQL bundle file could not be read")+    Right bytes+      | fromIntegral (ByteString.length bytes) /= byteLength step -> Left (simpleError VerifyBundle "SQL bundle file changed")+      | hex (SHA256.hash bytes) /= contentDigest step -> Left (simpleError VerifyBundle "SQL bundle digest changed")+      | otherwise -> Right bytes++nativeError :: LoadPhase -> Maybe FixtureName -> Maybe Int -> NativeError -> LoadError+nativeError phase fixture stepIndex NativeError {sqlState, reason} =+  LoadError {phase, targetIdentity = Nothing, fixture, stepIndex, cause = reason, sqlState, cleanupFailure = Nothing}++simpleError :: LoadPhase -> Text -> LoadError+simpleError phase cause = LoadError {phase, targetIdentity = Nothing, fixture = Nothing, stepIndex = Nothing, cause, sqlState = Nothing, cleanupFailure = Nothing}++sessionError :: SessionError -> LoadError+sessionError failure = simpleError BeginLoad (Text.pack (show failure))++hex :: ByteString.ByteString -> Text+hex = Text.pack . concatMap digits . ByteString.unpack+  where+    digits byte = case showHex byte "" of+      [digit] -> ['0', digit]+      value -> value
+ src/Hinagata/Postgres/Internal/Sql.hs view
@@ -0,0 +1,20 @@+-- | SQL helpers used by the native loader.+module Hinagata.Postgres.Internal.Sql+  ( setLocalDeadlines,+  )+where++import Data.ByteString.Char8 qualified as ByteString+import Database.PostgreSQL.LibPQ qualified as PQ+import Hinagata.Postgres.Error+import Hinagata.Postgres.Internal.Libpq++setLocalDeadlines :: PQ.Connection -> SessionOptions -> Deadline -> IO (Either NativeError ())+setLocalDeadlines connection options deadline =+  query connection deadline SetDeadline $+    ByteString.pack+      ( "SET LOCAL statement_timeout = "+          ++ show (statementDeadlineMs options)+          ++ "; SET LOCAL lock_timeout = "+          ++ show (lockDeadlineMs options)+      )
+ src/Hinagata/Postgres/Lease.hs view
@@ -0,0 +1,585 @@+-- | Bracket a single isolated clone of a sealed baseline.+module Hinagata.Postgres.Lease+  ( LeaseInfo (..),+    LeaseError (..),+    PreparationFailure (..),+    PreparationFailureKind (..),+    PreparationContext (..),+    RetentionPolicy (..),+    LeaseDisposition (..),+    LeaseTimings (..),+    LeaseOutcome (..),+    withDatabase,+    withDatabaseClassified,+    acquireDetached,+  )+where++import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (newEmptyMVar, readMVar, tryPutMVar)+import Control.Exception (Exception, IOException, SomeAsyncException, SomeException, evaluate, fromException, mask, throwIO, try)+import Data.Bifunctor (first)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Word (Word64)+import Database.PostgreSQL.LibPQ qualified as PQ+import GHC.Clock (getMonotonicTimeNSec)+import Hinagata.Config+import Hinagata.Connection (ConnectionTarget)+import Hinagata.Fixture.Bundle+import Hinagata.Postgres.Error (FixtureLoadContext, NativeError (..), NativePhase (..), SessionError, loadErrorContext)+import Hinagata.Postgres.Error qualified as Error+import Hinagata.Postgres.Internal.Access qualified as Access+import Hinagata.Postgres.Internal.BaselineRef+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Postgres.Internal.Load (loadComposedRemainder)+import Hinagata.Postgres.Ownership+import Hinagata.Prelude+import Hinagata.Types+import System.Timeout (timeout)+import Text.Read (readMaybe)++-- | Application credentials are held only in the target. Its Show instance+-- redacts the password; maintenance credentials never leave this module.+data LeaseInfo = LeaseInfo+  { leaseId :: !Text,+    runId :: !Text,+    applicationTarget :: !ConnectionTarget+  }+  deriving stock (Eq, Show)++-- | A lifecycle failure, distinct from a consumer callback's exception.+data LeaseError = LeaseError+  { cause :: !Text,+    sqlState :: !(Maybe Text),+    cleanupFailure :: !(Maybe Text),+    preparationContext :: !(Maybe PreparationContext),+    fixtureLoadContext :: !(Maybe FixtureLoadContext)+  }+  deriving stock (Eq, Show)++-- | Trusted adapters may raise only these credential-free outcomes. Legacy+-- callbacks returning Left Text retain their generic diagnostic.+data PreparationFailureKind+  = PreparationSpawnFailed+  | PreparationIdentityUnavailable+  | PreparationExit !Int+  | PreparationTimeout+  | PreparationCancelled !Int+  | PreparationCleanupFailed+  deriving stock (Eq, Show)++-- | A supervised preparation outcome, with an independent proof that releasing+-- the clone cannot race a still-running process. Unproven stoppage retains it.+data PreparationFailure = PreparationFailure+  { failureKind :: !PreparationFailureKind,+    releaseSafe :: !Bool+  }+  deriving stock (Eq, Show)+  deriving anyclass (Exception)++-- | Generated correlation identifiers attached after a clone has been allocated.+-- The context contains no endpoint, child output, arguments, or credentials.+data PreparationContext = PreparationContext+  { preparationFailure :: !PreparationFailure,+    preparationRunId :: !Text,+    preparationLeaseId :: !Text+  }+  deriving stock (Eq, Show)++-- | What to do after the consumer returns a classified failure or throws.+-- A detached acquisition always keeps its clone until explicit release.+data RetentionPolicy = ReleaseAlways | PreserveFailures | DetachAlways+  deriving stock (Eq, Show)++-- | Catalog state after the callback scope exits.+data LeaseDisposition = LeaseReleased | LeasePreserved | LeaseDetached | LeaseCleanupFailed+  deriving stock (Eq, Show)++-- | Monotonic stage durations for a completed clone scope. Queue time is+-- populated by the manager; the direct lease path reports zero for it.+data LeaseTimings = LeaseTimings+  { queueWaitMs :: !Int,+    catalogMs :: !Int,+    generationLockMs :: !Int,+    cloneMs :: !Int,+    scenarioLoadMs :: !Int,+    completionMs :: !Int+  }+  deriving stock (Eq, Show)++-- | Keep the callback value unchanged while reporting release or retention+-- separately. Setup failures still return 'Left' before a callback runs.+data LeaseOutcome a = LeaseOutcome+  { callbackValue :: !a,+    leaseInfo :: !LeaseInfo,+    leaseDisposition :: !LeaseDisposition,+    cleanupDiagnostic :: !(Maybe LeaseError),+    timings :: !LeaseTimings+  }+  deriving stock (Eq, Show)++data Clone = Clone+  { allocationId :: !Text,+    cloneLeaseId :: !Text,+    cloneRunId :: !Text,+    owned :: !OwnedDatabase+  }++-- | Allocate, load, hand off, and release one clone. The callback must close+-- its own pools and connections before returning. A failed setup or thrown+-- callback still attempts bounded, ownership-checked release. Ambiguous+-- allocations remain in the catalog for explicit recovery.+withDatabase :: HinagataConfig -> BaselineRef -> FixturePlan -> (LeaseInfo -> IO a) -> IO (Either LeaseError a)+withDatabase configuration baseline scenario callback = do+  completed <- withDatabaseClassified configuration baseline scenario ReleaseAlways (const False) callback+  pure $ case completed of+    Left problem -> Left problem+    Right LeaseOutcome {cleanupDiagnostic = Just problem} -> Left problem+    Right LeaseOutcome {callbackValue} -> Right callbackValue++-- | Classify a returned value without changing it. A classified failure is+-- preserved only under 'PreserveFailures'; synchronous callback exceptions+-- follow the same policy. Asynchronous cancellation instead attempts release+-- and rethrows the original exception, including cancellation during completion.+-- An interrupted completion may leave an inspectable orphan for explicit recovery.+withDatabaseClassified :: HinagataConfig -> BaselineRef -> FixturePlan -> RetentionPolicy -> (a -> Bool) -> (LeaseInfo -> IO a) -> IO (Either LeaseError (LeaseOutcome a))+withDatabaseClassified configuration baseline scenario policy classifyResult callback = do+  expiry <- acquisitionExpiry configuration+  timingRef <- newIORef emptyTimings+  case composePlans (baselinePlan baseline) scenario of+    Left _ -> pure (Left (failure "scenario conflicts with the frozen baseline"))+    Right composed -> do+      verified <- try @IOException (withinDeadline expiry (verifyComposedRemainder scenario composed))+      case verified of+        Left _ -> pure (Left (failure "scenario bundle could not be verified"))+        Right Nothing -> pure (Left acquisitionTimeout)+        Right (Just False) -> pure (Left (failure "scenario bundle changed"))+        Right (Just True) -> start timingRef expiry composed+  where+    options = defaultSessionOptions {operationDeadlineMs = positiveValue (acquisitionDeadlineMs configuration)}+    maintenance = first accessFailure (Access.adminTarget configuration (maintenanceDatabase configuration))++    start timingRef expiry composed = case maintenance of+      Left problem -> pure (Left problem)+      Right target -> do+        (timedCatalog, catalogDuration) <- measure (withinDeadline expiry (ensureCatalog target (maintenanceSchema configuration) options))+        modifyIORef' timingRef (\record -> record {catalogMs = catalogDuration})+        let catalog = maybe (Left (CatalogError "acquisition deadline expired" Nothing Nothing)) id timedCatalog+        case catalog of+          Left CatalogError {cause, sqlState} -> pure (Left (LeaseError cause sqlState Nothing Nothing Nothing))+          Right identity+            | identity /= clusterIdentity baseline -> pure (Left (failure "baseline belongs to a different catalog or cluster"))+            | otherwise -> do+                remaining <- remainingMs expiry+                case remaining of+                  Nothing -> pure (Left acquisitionTimeout)+                  Just milliseconds -> do+                    let sessionOptions = options {connectDeadlineMs = min (connectDeadlineMs options) milliseconds}+                    opened <- withSession target sessionOptions $ \session -> mask $ \restore -> do+                      acquired <- restore (withinDeadline expiry (allocate timingRef configuration baseline session target options identity))+                      case acquired of+                        Nothing -> pure (Left acquisitionTimeout)+                        Just (Left problem) -> pure (Left problem)+                        Just (Right clone) -> do+                          inHook <- newIORef False+                          prepared <- try @SomeException (restore (withinDeadline expiry (prepareClone inHook timingRef session clone composed)))+                          case prepared of+                            Left exception -> case fromException @PreparationFailure exception of+                              Just detail -> do+                                let problem = preparationError clone detail+                                cleaned <-+                                  try @SomeException $+                                    if releaseSafe detail+                                      then release configuration session target options identity clone+                                      else markStateWithDiagnostic configuration session clone "Preserved" (Just "clone preparation process cleanup uncertain; establish process stoppage before release")+                                pure (mergeCleanup (Left problem) cleaned)+                              Nothing -> do+                                _ <- try @SomeException (release configuration session target options identity clone)+                                throwIO exception+                            Right Nothing -> do+                              hookActive <- readIORef inHook+                              let problem = if hookActive then preparationError clone (PreparationFailure PreparationTimeout True) else acquisitionTimeout+                              cleaned <- try @SomeException (release configuration session target options identity clone)+                              pure (mergeCleanup (Left problem) cleaned)+                            Right (Just (Left problem)) -> do+                              cleaned <- try @SomeException (release configuration session target options identity clone)+                              pure (mergeCleanup (Left problem) cleaned)+                            Right (Just (Right info)) -> do+                              returned <-+                                try @SomeException+                                  ( restore+                                      ( raceWithOwnership+                                          session+                                          ( do+                                              value <- callback info+                                              failed <- if policy == DetachAlways then pure False else evaluate (classifyResult value)+                                              pure (value, failed)+                                          )+                                      )+                                  )+                              case returned of+                                Left exception -> do+                                  _ <- try @SomeException (completeFailure (isJust (fromException @SomeAsyncException exception)) session target identity clone)+                                  throwIO exception+                                Right (Right problem) -> pure (Left problem)+                                Right (Left (value, failed)) -> do+                                  (finished, duration) <- measure (try @SomeException (completeValue session target identity clone failed))+                                  modifyIORef' timingRef (\record -> record {completionMs = duration})+                                  recorded <- readIORef timingRef+                                  case finished of+                                    Left exception | isJust (fromException @SomeAsyncException exception) -> throwIO exception+                                    _ -> pure (Right (outcomeFor recorded info value finished))+                    pure $ case opened of+                      Left problem -> Left (sessionFailure problem)+                      Right result -> result++    completeFailure cancelled session target identity clone+      | cancelled = release configuration session target options identity clone+      | otherwise = case policy of+          ReleaseAlways -> release configuration session target options identity clone+          PreserveFailures -> markState configuration session clone "Preserved"+          DetachAlways -> markState configuration session clone "Detached"++    completeValue session target identity clone failed+      | policy == DetachAlways = fmap (\result -> (LeaseDetached, result)) (markState configuration session clone "Detached")+      | policy == PreserveFailures && failed = fmap (\result -> (LeasePreserved, result)) (markState configuration session clone "Preserved")+      | otherwise = fmap (\result -> (LeaseReleased, result)) (release configuration session target options identity clone)++    outcomeFor recorded info value = \case+      Left _ -> LeaseOutcome value info LeaseCleanupFailed (Just (failure "lease completion raised an exception")) recorded+      Right (_, Left problem) -> LeaseOutcome value info LeaseCleanupFailed (Just problem) recorded+      Right (disposition, Right ()) -> LeaseOutcome value info disposition Nothing recorded++    raceWithOwnership session action = mask $ \restore -> do+      stop <- newEmptyMVar+      watcher <- Async.async (watchOwnership session stop)+      raced <- try @SomeException (restore (Async.race action (Async.wait watcher)))+      _ <- tryPutMVar stop ()+      -- Async.race cancels only the thread waiting on watcher. Let an in-flight+      -- ping finish before release uses the same maintenance session.+      settled <- Async.waitCatch watcher+      case raced of+        Left exception -> throwIO exception+        Right (Right (Just problem)) -> pure (Right problem)+        Right (Right Nothing) -> pure (Right (failure "ownership watcher stopped before callback"))+        Right (Left value) -> case settled of+          Right Nothing -> pure (Left value)+          Right (Just problem) -> pure (Right problem)+          Left exception -> throwIO exception++    watchOwnership session stop = do+      stopped <- timeout 250000 (readMVar stop)+      case stopped of+        Just () -> pure Nothing+        Nothing -> do+          checked <- runExclusive session $ \connection _ -> do+            deadline <- deadlineAfter 1000+            ping <- query connection deadline Sql "SELECT 1"+            pure $ case ping of+              Right () -> Keep True+              Left _ -> Retire False+          case checked of+            Right True -> watchOwnership session stop+            _ -> pure (Just (failure "lease ownership connection was lost during callback"))++    prepareClone inHook timingRef session clone composed = do+      let OwnedDatabase {name = database} = owned clone+      prepared <- runExclusive session $ \connection sessionOptions -> do+        deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+        result <- Access.prepareAccess connection sessionOptions configuration database deadline+        pure (Keep (first accessFailure result))+      case flatten prepared of+        Left problem -> pure (Left problem)+        Right () -> case first accessFailure (Access.accessToTarget configuration (setup configuration) database) of+          Left problem -> pure (Left problem)+          Right setupTarget -> do+            preparedHook <- case cloneHook baseline of+              Nothing -> pure (Right ())+              Just hook -> do+                writeIORef inHook True+                result <- hook setupTarget+                writeIORef inHook False+                pure $ case result of Left _ -> Left (failure "clone preparation hook failed"); Right () -> Right ()+            case preparedHook of+              Left problem -> pure (Left problem)+              Right () -> do+                (loaded, duration) <- measure (withSession setupTarget options (\setupSession -> loadComposedRemainder setupSession scenario composed))+                modifyIORef' timingRef (\record -> record {scenarioLoadMs = duration})+                case loaded of+                  Left problem -> pure (Left (sessionFailure problem))+                  Right (Left problem@Error.LoadError {Error.sqlState}) -> pure (Left (LeaseError "scenario fixture load failed" sqlState Nothing Nothing (Just (loadErrorContext problem))))+                  Right (Right _) -> do+                    active <- markState configuration session clone "Active"+                    case active of+                      Left problem -> pure (Left problem)+                      Right () -> case first accessFailure (Access.accessToTarget configuration (application configuration) database) of+                        Left problem -> pure (Left problem)+                        Right appTarget -> pure (Right LeaseInfo {leaseId = cloneLeaseId clone, runId = cloneRunId clone, applicationTarget = appTarget})++-- | Acquire and retain a clone without a callback scope. It is recorded as+-- Detached and needs an explicit release by lease ID through cleanup.+acquireDetached :: HinagataConfig -> BaselineRef -> FixturePlan -> IO (Either LeaseError LeaseInfo)+acquireDetached configuration baseline scenario = do+  completed <- withDatabaseClassified configuration baseline scenario DetachAlways (const False) pure+  pure $ case completed of+    Left problem -> Left problem+    Right LeaseOutcome {cleanupDiagnostic = Just problem} -> Left problem+    Right LeaseOutcome {leaseInfo} -> Right leaseInfo++allocate :: IORef LeaseTimings -> HinagataConfig -> BaselineRef -> Session -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> IO (Either LeaseError Clone)+allocate timingRef configuration baseline session maintenance options identity = do+  acquired <- runExclusive session $ \connection sessionOptions -> do+    deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+    let lockKey = generationLockKey baseline+    (locked, duration) <- measure (queryParamRows connection deadline Sql "SELECT pg_advisory_lock_shared(1212761905, hashtext($1))" [Just (Encoding.encodeUtf8 lockKey)] 1)+    modifyIORef' timingRef (\record -> record {generationLockMs = duration})+    case locked of+      Left problem -> pure (Keep (Left (nativeFailure problem)))+      Right _ -> do+        let generations = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"generations\""+            statement = Encoding.encodeUtf8 ("SELECT database_name::text, database_oid::text, ownership_token::text, state FROM " <> generations <> " WHERE id = $1 AND project_id = $2")+            args = map (Just . Encoding.encodeUtf8) [generationId baseline, projectIdText (project configuration)]+        rows <- queryParamRows connection deadline Sql statement args 1+        case rows of+          Left problem -> pure (Keep (Left (nativeFailure problem)))+          Right [[Just database, Just oidBytes, Just tokenBytes, Just "Ready"]]+            | database == Encoding.encodeUtf8 (databaseNameText (baselineName baseline))+                && oidBytes == Encoding.encodeUtf8 (Text.pack (show (oid (ownership baseline))))+                && tokenBytes == Encoding.encodeUtf8 (token (ownership baseline)) -> do+                sealed <- queryParamRows connection deadline Sql "SELECT datallowconn::text FROM pg_database WHERE datname = $1" [Just database] 1+                pure (Keep (case sealed of Right [[Just "false"]] -> Right (); Left problem -> Left (nativeFailure problem); _ -> Left (failure "baseline is not sealed")))+          Right _ -> pure (Keep (Left (failure "baseline generation is no longer ready or has changed identity")))+  case flatten acquired of+    Left problem -> pure (Left problem)+    Right () -> do+      evidence <- verifyOwnedDatabase maintenance options identity (ownership baseline)+      case evidence of+        Right OwnershipMatches ->+          flatten+            <$> runExclusive+              session+              ( \connection sessionOptions -> do+                  (created, duration) <- measure (createClone configuration baseline connection sessionOptions identity)+                  modifyIORef' timingRef (\record -> record {cloneMs = duration})+                  pure (Keep created)+              )+        Right _ -> pure (Left (failure "baseline lost positive ownership"))+        Left CatalogError {cause, sqlState} -> pure (Left (LeaseError cause sqlState Nothing Nothing Nothing))++createClone :: HinagataConfig -> BaselineRef -> PQ.Connection -> SessionOptions -> CatalogIdentity -> IO (Either LeaseError Clone)+createClone configuration baseline connection sessionOptions identity = do+  deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+  ids <- queryParamRows connection deadline Sql "SELECT gen_random_uuid()::text, gen_random_uuid()::text, gen_random_uuid()::text, gen_random_uuid()::text" [] 1+  case ids of+    Right [[Just allocationBytes, Just leaseBytes, Just runBytes, Just tokenBytes]] -> do+      let allocation = Encoding.decodeUtf8 allocationBytes+          lease = Encoding.decodeUtf8 leaseBytes+          run = Encoding.decodeUtf8 runBytes+          tokenText = Encoding.decodeUtf8 tokenBytes+          databaseText = "hl_" <> Text.take 8 (baselineFingerprint baseline) <> "_" <> Text.take 20 (Text.filter (/= '-') tokenText)+      case mkDatabaseName databaseText of+        Left _ -> pure (Left (failure "generated clone name is invalid"))+        Right database -> do+          let table = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"allocations\""+              insertSql = Encoding.encodeUtf8 ("INSERT INTO " <> table <> " (id, project_id, generation_id, run_id, database_name, ownership_token, state) VALUES ($1, $2, $3, $4, $5, $6::uuid, 'Allocating')")+              args = map (Just . Encoding.encodeUtf8) [allocation, projectIdText (project configuration), generationId baseline, run, databaseText, tokenText]+          inserted <- queryParamRows connection deadline Sql insertSql args 0+          case inserted of+            Left problem -> pure (Left (nativeFailure problem))+            Right _ -> do+              let strategy = case cloneStrategy configuration of WalLog -> "WAL_LOG"; FileCopy -> "FILE_COPY"+                  quoted = Access.quoteDatabase database+                  createSql = Encoding.encodeUtf8 ("CREATE DATABASE " <> quoted <> " TEMPLATE " <> Access.quoteDatabase (baselineName baseline) <> " STRATEGY " <> strategy <> " ALLOW_CONNECTIONS true")+              created <- query connection deadline Sql createSql+              case created of+                Left problem -> pure (Left (nativeFailure problem))+                Right () -> do+                  let marker = "hinagata:v1:" <> clusterUuid identity <> ":" <> tokenText+                  commented <- query connection deadline Sql (Encoding.encodeUtf8 ("COMMENT ON DATABASE " <> quoted <> " IS " <> Access.quoteLiteral marker))+                  case commented of+                    Left problem -> pure (Left (nativeFailure problem))+                    Right () -> do+                      observed <- queryParamRows connection deadline Sql "SELECT oid::text FROM pg_database WHERE datname = $1" [Just (Encoding.encodeUtf8 databaseText)] 1+                      case observed of+                        Right [[Just numberBytes]] -> case readMaybe (Text.unpack (Encoding.decodeUtf8 numberBytes)) of+                          Nothing -> pure (Left (failure "clone has an invalid OID"))+                          Just number -> do+                            let bindSql = Encoding.encodeUtf8 ("UPDATE " <> table <> " SET database_oid = $2::oid, state = 'Loading', updated_at = clock_timestamp() WHERE id = $1 AND state = 'Allocating' RETURNING id")+                            bound <- queryParamRows connection deadline Sql bindSql [Just allocationBytes, Just numberBytes] 1+                            case bound of+                              Right [[Just _]] -> do+                                let leases = quoteSqlIdentifier (maintenanceSchema configuration) <> ".\"leases\""+                                    leaseSql = Encoding.encodeUtf8 ("INSERT INTO " <> leases <> " (id, allocation_id, run_id, state) VALUES ($1, $2, $3, 'Loading')")+                                recorded <- queryParamRows connection deadline Sql leaseSql (map Just [leaseBytes, allocationBytes, runBytes]) 0+                                case recorded of+                                  Left problem -> pure (Left (nativeFailure problem))+                                  Right _ -> do+                                    let leaseKey = leaseLockKey lease+                                    guarded <- queryParamRows connection deadline Sql "SELECT pg_advisory_lock(1212761905, hashtext($1))" [Just (Encoding.encodeUtf8 leaseKey)] 1+                                    case guarded of+                                      Left problem -> pure (Left (nativeFailure problem))+                                      Right _ -> do+                                        unlocked <- queryParamRows connection deadline Sql "SELECT pg_advisory_unlock_shared(1212761905, hashtext($1))::text" [Just (Encoding.encodeUtf8 (generationLockKey baseline))] 1+                                        pure $ case unlocked of+                                          Right [[Just "true"]] -> Right (Clone allocation lease run (OwnedDatabase database number tokenText))+                                          Left problem -> Left (nativeFailure problem)+                                          Right _ -> Left (failure "generation allocation lock could not be released")+                              Left problem -> pure (Left (nativeFailure problem))+                              Right _ -> pure (Left (failure "clone identity could not be bound"))+                        Left problem -> pure (Left (nativeFailure problem))+                        Right _ -> pure (Left (failure "created clone is absent from the catalog"))+    Left problem -> pure (Left (nativeFailure problem))+    Right _ -> pure (Left (failure "clone UUID generation returned an invalid result"))++markState :: HinagataConfig -> Session -> Clone -> Text -> IO (Either LeaseError ())+markState configuration session clone state = markStateWithDiagnostic configuration session clone state Nothing++markStateWithDiagnostic :: HinagataConfig -> Session -> Clone -> Text -> Maybe Text -> IO (Either LeaseError ())+markStateWithDiagnostic configuration session clone state diagnostic =+  flatten+    <$> runExclusive+      session+      ( \connection options -> do+          deadline <- deadlineAfter (operationDeadlineMs options)+          let schema = quoteSqlIdentifier (maintenanceSchema configuration)+              allocationSql = Encoding.encodeUtf8 ("UPDATE " <> schema <> ".\"allocations\" SET state = $2, last_error = $3, updated_at = clock_timestamp() WHERE id = $1 RETURNING id")+              leaseSql = Encoding.encodeUtf8 ("UPDATE " <> schema <> ".\"leases\" SET state = $2, updated_at = clock_timestamp() WHERE id = $1 RETURNING id")+              update statement identifier = queryParamRows connection deadline Sql statement (map (Just . Encoding.encodeUtf8) [identifier, state]) 1+              rollback problem = do+                cleanupDeadline <- deadlineAfter (cleanupDeadlineMs options)+                rolled <- query connection cleanupDeadline Rollback "ROLLBACK"+                pure $ case rolled of+                  Right () -> Keep (Left problem)+                  Left _ -> Retire (Left (withCleanupDiagnostic problem "catalog state rollback failed"))+          begun <- query connection deadline Begin "BEGIN"+          case begun of+            Left problem -> pure (Retire (Left (nativeFailure problem)))+            Right () -> do+              allocation <- queryParamRows connection deadline Sql allocationSql [Just (Encoding.encodeUtf8 (allocationId clone)), Just (Encoding.encodeUtf8 state), Encoding.encodeUtf8 . Text.take 512 <$> diagnostic] 1+              case allocation of+                Right [[Just _]] -> do+                  lease <- update leaseSql (cloneLeaseId clone)+                  case lease of+                    Right [[Just _]] -> do+                      committed <- query connection deadline Commit "COMMIT"+                      pure $ case committed of+                        Right () -> Keep (Right ())+                        Left problem -> Retire (Left (nativeFailure problem))+                    Left problem -> rollback (nativeFailure problem)+                    Right _ -> rollback (failure "lease state record is missing")+                Left problem -> rollback (nativeFailure problem)+                Right _ -> rollback (failure "allocation state record is missing")+      )++release :: HinagataConfig -> Session -> ConnectionTarget -> SessionOptions -> CatalogIdentity -> Clone -> IO (Either LeaseError ())+release configuration session maintenance options identity clone = do+  evidence <- verifyOwnedDatabase maintenance options identity (owned clone)+  case evidence of+    Left CatalogError {cause, sqlState} -> pure (Left (LeaseError cause sqlState Nothing Nothing Nothing))+    Right OwnershipMismatch -> pure (Left (failure "clone ownership changed; release refused"))+    Right OwnershipMissing -> markState configuration session clone "Released"+    Right OwnershipMatches -> do+      releasing <- markState configuration session clone "Releasing"+      case releasing of+        Left problem -> pure (Left problem)+        Right () -> do+          dropped <-+            flatten+              <$> runExclusive+                session+                ( \connection sessionOptions -> do+                    deadline <- deadlineAfter (cleanupDeadlineMs sessionOptions)+                    let OwnedDatabase {name = database} = owned clone+                    outcome <- query connection deadline Sql (Encoding.encodeUtf8 ("DROP DATABASE " <> Access.quoteDatabase database <> " WITH (FORCE)"))+                    pure (Keep (first nativeFailure outcome))+                )+          case dropped of+            Left problem@LeaseError {cause = dropCause, sqlState = dropState} -> do+              marked <- markStateWithDiagnostic configuration session clone "CleanupFailed" (Just dropCause)+              pure $ case marked of+                Right () -> Left problem+                Left LeaseError {cause = recordCause} -> Left (LeaseError dropCause dropState (Just ("could not record cleanup failure: " <> recordCause)) Nothing Nothing)+            Right () -> markState configuration session clone "Released"++mergeCleanup :: Either LeaseError a -> Either SomeException (Either LeaseError ()) -> Either LeaseError a+mergeCleanup result cleanup = case cleanup of+  Left _ -> case result of+    Left problem -> Left (withCleanupDiagnostic problem "clone release raised an exception")+    Right _ -> Left (failure "clone release raised an exception")+  Right (Left problem@LeaseError {cause = cleanupCause}) -> case result of+    Left primary -> Left (withCleanupDiagnostic primary cleanupCause)+    Right _ -> Left problem+  Right (Right ()) -> result++flatten :: Either SessionError (Either LeaseError a) -> Either LeaseError a+flatten = either (Left . sessionFailure) id++failure :: Text -> LeaseError+failure cause = LeaseError cause Nothing Nothing Nothing Nothing++nativeFailure :: NativeError -> LeaseError+nativeFailure NativeError {reason, sqlState} = LeaseError reason sqlState Nothing Nothing Nothing++accessFailure :: Access.AccessError -> LeaseError+accessFailure Access.AccessError {cause, sqlState} = LeaseError cause sqlState Nothing Nothing Nothing++sessionFailure :: SessionError -> LeaseError+sessionFailure problem = failure (Text.pack (show problem))++generationLockKey :: BaselineRef -> Text+generationLockKey baseline = "generation:" <> generationId baseline++leaseLockKey :: Text -> Text+leaseLockKey lease = "lease:" <> lease++acquisitionTimeout :: LeaseError+acquisitionTimeout = failure "acquisition deadline expired"++emptyTimings :: LeaseTimings+emptyTimings = LeaseTimings 0 0 0 0 0 0++measure :: IO a -> IO (a, Int)+measure action = do+  started <- getMonotonicTimeNSec+  result <- action+  finished <- getMonotonicTimeNSec+  pure (result, fromIntegral ((finished - started) `div` 1000000))++acquisitionExpiry :: HinagataConfig -> IO Word64+acquisitionExpiry configuration = do+  now <- getMonotonicTimeNSec+  pure (now + fromIntegral (positiveValue (acquisitionDeadlineMs configuration)) * 1000000)++remainingMs :: Word64 -> IO (Maybe Int)+remainingMs expiry = do+  now <- getMonotonicTimeNSec+  pure $ if now >= expiry then Nothing else Just (fromIntegral ((expiry - now) `div` 1000000) `max` 1)++withinDeadline :: Word64 -> IO a -> IO (Maybe a)+withinDeadline expiry action = do+  now <- getMonotonicTimeNSec+  if now >= expiry+    then pure Nothing+    else timeout (max 1 (fromIntegral (min (fromIntegral (maxBound :: Int)) ((expiry - now) `div` 1000)))) action++preparationError :: Clone -> PreparationFailure -> LeaseError+preparationError clone detail =+  LeaseError+    (reason (failureKind detail))+    Nothing+    (if releaseSafe detail then Nothing else Just "child process cleanup uncertain; clone preserved")+    (Just (PreparationContext detail (cloneRunId clone) (cloneLeaseId clone)))+    Nothing+  where+    reason = \case+      PreparationSpawnFailed -> "clone preparation command could not be started"+      PreparationIdentityUnavailable -> "clone preparation process identity unavailable"+      PreparationExit status -> "clone preparation command exited with status " <> Text.pack (show status)+      PreparationTimeout -> "clone preparation deadline expired"+      PreparationCancelled signal -> "clone preparation cancelled by signal " <> Text.pack (show signal)+      PreparationCleanupFailed -> "clone preparation process cleanup could not be proven"++withCleanupDiagnostic :: LeaseError -> Text -> LeaseError+withCleanupDiagnostic LeaseError {cause, sqlState, preparationContext, fixtureLoadContext} diagnostic = LeaseError cause sqlState (Just diagnostic) preparationContext fixtureLoadContext
+ src/Hinagata/Postgres/Load.hs view
@@ -0,0 +1,9 @@+-- | Atomic direct loading into a caller-selected, already migrated database.+module Hinagata.Postgres.Load+  ( LoadReport (..),+    loadPlan,+    loadInto,+  )+where++import Hinagata.Postgres.Internal.Load (LoadReport (..), loadInto, loadPlan)
+ src/Hinagata/Postgres/Manager.hs view
@@ -0,0 +1,250 @@+-- | Manager-local admission for active clones and setup workers.+module Hinagata.Postgres.Manager+  ( Manager,+    DatabaseRequest (..),+    withManager,+    withManagedDatabase,+    withManagedDatabaseClassified,+    acquireManagedDetached,+    withDatabases,+  )+where++import Control.Concurrent.MVar+import Control.Exception (bracket, finally, mask, onException)+import Data.Foldable (traverse_)+import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef)+import Data.List (sortOn)+import Data.List.NonEmpty qualified as NonEmpty+import Data.Text qualified as Text+import Data.Word (Word64)+import GHC.Clock (getMonotonicTimeNSec)+import Hinagata.Config+import Hinagata.Fixture.Bundle (FixturePlan)+import Hinagata.Postgres.Baseline (BaselineRef)+import Hinagata.Postgres.Lease (LeaseError (..), LeaseInfo, LeaseOutcome (..), LeaseTimings (..), RetentionPolicy (..), withDatabase, withDatabaseClassified)+import Hinagata.Prelude+import Hinagata.Types (mkPositive, positiveValue)+import System.Timeout (timeout)++-- | A scoped, local admission controller. It does not enforce a cluster-wide+-- connection budget; callers must allow for other managers and service pools.+data Manager = Manager+  { configuration :: !HinagataConfig,+    activeGate :: !Gate,+    setupGate :: !Gate+  }++-- | One named clone request. Names must be unique within a collection.+data DatabaseRequest = DatabaseRequest+  { requestName :: !Text,+    baseline :: !BaselineRef,+    scenario :: !FixturePlan+  }++data Gate = Gate+  { limit :: !Int,+    pendingLimit :: !Int,+    state :: !(MVar GateState)+  }++data Waiter = Waiter+  { ticket :: !Int,+    requested :: !Int,+    signal :: !(MVar Bool)+  }++data GateState = GateState+  { used :: !Int,+    queued :: ![Waiter],+    granted :: ![(Int, Int)],+    nextTicket :: !Int,+    closed :: !Bool+  }++data Admission = Immediate | Queued !Int !(MVar Bool)++-- | Bracket manager-local capacity. Closing a manager wakes queued callers+-- with a typed error; callers must finish active callbacks before scope exit.+withManager :: HinagataConfig -> (Manager -> IO a) -> IO a+withManager configuration = bracket create closeManager+  where+    create = do+      activeGate <- newGate (positiveValue (activeLeases configuration)) (positiveValue (pendingRequests configuration))+      setupGate <- newGate (positiveValue (setupWorkers configuration)) (positiveValue (activeLeases configuration))+      pure Manager {configuration, activeGate, setupGate}++-- | Reserve one active slot, then a setup worker until callback handoff.+-- Queue saturation is reported before allocating any database.+withManagedDatabase :: Manager -> BaselineRef -> FixturePlan -> (LeaseInfo -> IO a) -> IO (Either LeaseError a)+withManagedDatabase manager baseline scenario callback = do+  expiry <- acquisitionExpiry manager+  withPermitUntil (activeGate manager) 1 expiry (withSetupUntil manager expiry baseline scenario callback)++-- | Apply the same admission limits to a result-classified callback.+withManagedDatabaseClassified :: Manager -> BaselineRef -> FixturePlan -> RetentionPolicy -> (a -> Bool) -> (LeaseInfo -> IO a) -> IO (Either LeaseError (LeaseOutcome a))+withManagedDatabaseClassified manager baseline scenario policy classifyResult callback = do+  expiry <- acquisitionExpiry manager+  started <- getMonotonicTimeNSec+  queueRef <- newIORef 0+  completed <- withPermitUntil (activeGate manager) 1 expiry $ do+    activeAdmitted <- getMonotonicTimeNSec+    modifyIORef' queueRef (+ millisecondsBetween started activeAdmitted)+    withSetupUsing manager expiry (Just queueRef) (\configuration handoff -> withDatabaseClassified configuration baseline scenario policy classifyResult handoff) callback+  waited <- readIORef queueRef+  pure $ fmap (\outcome@LeaseOutcome {timings} -> outcome {timings = timings {queueWaitMs = waited}}) completed++-- | Acquire a detached clone through manager admission, freeing active+-- capacity once its retained identity has been recorded.+acquireManagedDetached :: Manager -> BaselineRef -> FixturePlan -> IO (Either LeaseError LeaseInfo)+acquireManagedDetached manager baseline scenario = do+  completed <- withManagedDatabaseClassified manager baseline scenario DetachAlways (const False) pure+  pure $ case completed of+    Left problem -> Left problem+    Right LeaseOutcome {cleanupDiagnostic = Just problem} -> Left problem+    Right LeaseOutcome {leaseInfo} -> Right leaseInfo++-- | Reserve capacity for the whole named set, acquire in name order, and+-- unwind earlier clones if any later acquisition fails. The callback receives+-- names paired with application endpoints in that same stable order.+withDatabases :: Manager -> NonEmpty DatabaseRequest -> ([(Text, LeaseInfo)] -> IO a) -> IO (Either LeaseError a)+withDatabases manager requests callback = do+  let ordered = sortOn requestName (NonEmpty.toList requests)+      count = length ordered+  if any (Text.null . requestName) ordered || duplicateNames ordered+    then pure (Left (admissionError "database request names must be nonempty and unique"))+    else do+      expiry <- acquisitionExpiry manager+      withPermitUntil (activeGate manager) count expiry (acquireAll expiry ordered [] callback)+  where+    acquireAll _ [] accumulated callbackAction = Right <$> callbackAction (reverse accumulated)+    acquireAll expiry (DatabaseRequest {requestName, baseline, scenario} : rest) accumulated callbackAction = do+      result <- withSetupUntil manager expiry baseline scenario (\info -> acquireAll expiry rest ((requestName, info) : accumulated) callbackAction)+      pure (result >>= id)++    duplicateNames (left : right : rest) = requestName left == requestName right || duplicateNames (right : rest)+    duplicateNames _ = False++withSetupUntil :: Manager -> Word64 -> BaselineRef -> FixturePlan -> (LeaseInfo -> IO a) -> IO (Either LeaseError a)+withSetupUntil manager expiry baseline scenario =+  withSetupUsing manager expiry Nothing (\configuration handoff -> withDatabase configuration baseline scenario handoff)++withSetupUsing :: Manager -> Word64 -> Maybe (IORef Int) -> (HinagataConfig -> (LeaseInfo -> IO a) -> IO (Either LeaseError b)) -> (LeaseInfo -> IO a) -> IO (Either LeaseError b)+withSetupUsing Manager {configuration, setupGate} expiry waitingRef runner callback = mask $ \restore -> do+  waitingSince <- getMonotonicTimeNSec+  admitted <- acquireUntil setupGate 1 expiry+  admittedAt <- getMonotonicTimeNSec+  traverse_ (\reference -> modifyIORef' reference (+ millisecondsBetween waitingSince admittedAt)) waitingRef+  case admitted of+    Left problem -> pure (Left problem)+    Right () -> do+      released <- newIORef False+      let releaseOnce = do+            shouldRelease <- atomicModifyIORef' released (\done -> (True, not done))+            when shouldRelease (releaseGate setupGate 1)+      remaining <- remainingMs expiry+      case remaining of+        Nothing -> releaseOnce >> pure (Left (admissionError "acquisition deadline expired during setup admission"))+        Just milliseconds -> case mkPositive "acquisition deadline" milliseconds of+          Left _ -> releaseOnce >> pure (Left (admissionError "acquisition deadline is invalid"))+          Right budget ->+            restore (runner (configuration {acquisitionDeadlineMs = budget}) (\info -> releaseOnce >> callback info)) `finally` releaseOnce++withPermitUntil :: Gate -> Int -> Word64 -> IO (Either LeaseError a) -> IO (Either LeaseError a)+withPermitUntil gate amount expiry action = mask $ \restore -> do+  admitted <- acquireUntil gate amount expiry+  case admitted of+    Left problem -> pure (Left problem)+    Right () -> restore action `finally` releaseGate gate amount++acquisitionExpiry :: Manager -> IO Word64+acquisitionExpiry Manager {configuration} = do+  now <- getMonotonicTimeNSec+  pure (now + fromIntegral (positiveValue (acquisitionDeadlineMs configuration)) * 1000000)++remainingMs :: Word64 -> IO (Maybe Int)+remainingMs expiry = do+  now <- getMonotonicTimeNSec+  pure $ if now >= expiry then Nothing else Just (fromIntegral ((expiry - now) `div` 1000000) `max` 1)++millisecondsBetween :: Word64 -> Word64 -> Int+millisecondsBetween start finish = fromIntegral ((finish - start) `div` 1000000)++acquireUntil :: Gate -> Int -> Word64 -> IO (Either LeaseError ())+acquireUntil gate amount expiry = do+  now <- getMonotonicTimeNSec+  if now >= expiry+    then pure (Left (admissionError "acquisition deadline expired"))+    else do+      let microseconds = fromIntegral (min (fromIntegral (maxBound :: Int)) ((expiry - now) `div` 1000))+      outcome <- timeout (max 1 microseconds) (acquireGate gate amount)+      pure (fromMaybe (Left (admissionError "acquisition deadline expired while queued")) outcome)++newGate :: Int -> Int -> IO Gate+newGate limit pendingLimit = do+  state <- newMVar GateState {used = 0, queued = [], granted = [], nextTicket = 0, closed = False}+  pure Gate {limit, pendingLimit, state}++acquireGate :: Gate -> Int -> IO (Either LeaseError ())+acquireGate gate amount = mask $ \restore -> do+  signal <- newEmptyMVar+  decision <- modifyMVar (state gate) $ \current ->+    if closed current+      then pure (current, Left (admissionError "manager is closed"))+      else+        if amount > limit gate+          then pure (current, Left (admissionError "collection exceeds the active-lease limit"))+          else+            if null (queued current) && used current + amount <= limit gate+              then pure (current {used = used current + amount}, Right Immediate)+              else+                if length (queued current) >= pendingLimit gate+                  then pure (current, Left (admissionError "manager pending-request limit is full"))+                  else do+                    let identifier = nextTicket current+                        waiter = Waiter identifier amount signal+                    pure (current {queued = queued current ++ [waiter], nextTicket = identifier + 1}, Right (Queued identifier signal))+  case decision of+    Left problem -> pure (Left problem)+    Right Immediate -> pure (Right ())+    Right (Queued identifier waiting) -> do+      accepted <- restore (takeMVar waiting) `onException` cancelWaiter gate identifier+      if accepted+        then acknowledge gate identifier >> pure (Right ())+        else pure (Left (admissionError "manager closed while request was queued"))++acknowledge :: Gate -> Int -> IO ()+acknowledge gate identifier = modifyMVar_ (state gate) $ \current ->+  pure current {granted = filter ((/= identifier) . fst) (granted current)}++cancelWaiter :: Gate -> Int -> IO ()+cancelWaiter gate identifier = modifyMVar_ (state gate) $ \current -> do+  let waiting = filter ((/= identifier) . ticket) (queued current)+      reserved = lookup identifier (granted current)+      withoutGrant = filter ((/= identifier) . fst) (granted current)+      released = current {queued = waiting, granted = withoutGrant, used = used current - maybe 0 id reserved}+  wake gate released++releaseGate :: Gate -> Int -> IO ()+releaseGate gate amount = modifyMVar_ (state gate) $ \current ->+  wake gate current {used = used current - amount}++wake :: Gate -> GateState -> IO GateState+wake gate current = case queued current of+  [] -> pure current+  waiter : rest+    | used current + requested waiter <= limit gate && not (closed current) -> do+        putMVar (signal waiter) True+        wake gate current {used = used current + requested waiter, queued = rest, granted = (ticket waiter, requested waiter) : granted current}+    | otherwise -> pure current++closeManager :: Manager -> IO ()+closeManager Manager {activeGate, setupGate} = closeGate activeGate >> closeGate setupGate++closeGate :: Gate -> IO ()+closeGate gate = modifyMVar_ (state gate) $ \current -> do+  traverse_ (\waiter -> putMVar (signal waiter) False) (queued current)+  pure current {queued = [], closed = True}++admissionError :: Text -> LeaseError+admissionError cause = LeaseError cause Nothing Nothing Nothing Nothing
+ src/Hinagata/Postgres/Ownership.hs view
@@ -0,0 +1,225 @@+-- | Versioned maintenance-catalog bootstrap on a caller-selected database.+module Hinagata.Postgres.Ownership+  ( CatalogIdentity (..),+    CatalogError (..),+    OwnedDatabase (..),+    OwnershipEvidence (..),+    ensureCatalog,+    inspectCatalog,+    verifyOwnedDatabase,+  )+where++import Control.Exception (IOException, try)+import Data.ByteString qualified as ByteString+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Database.PostgreSQL.LibPQ qualified as PQ+import Hinagata.Connection (ConnectionTarget)+import Hinagata.Postgres.Error (NativeError (..), NativePhase (..), SessionError (..))+import Hinagata.Postgres.Internal.Libpq+import Hinagata.Prelude+import Hinagata.Types (DatabaseName, SqlIdentifier, databaseNameText, quoteSqlIdentifier, sqlIdentifierText)+import Paths_hinagata_postgres (getDataFileName)++-- | The catalog format and cluster marker read back from the maintenance+-- database. The marker prevents records from one cluster being used in another.+data CatalogIdentity = CatalogIdentity+  { formatVersion :: !Int,+    clusterUuid :: !Text+  }+  deriving stock (Eq, Show)++-- | Expected bootstrap failures. Raw server messages are not retained.+data CatalogError = CatalogError+  { cause :: !Text,+    sqlState :: !(Maybe Text),+    cleanupFailure :: !(Maybe Text)+  }+  deriving stock (Eq, Show)++-- | Identity recorded when Hinagata allocated a database. A name alone is+-- never sufficient authority for cleanup.+data OwnedDatabase = OwnedDatabase+  { name :: !DatabaseName,+    oid :: !Int,+    token :: !Text+  }+  deriving stock (Eq, Show)++-- | Comparison against the current cluster catalog and database comment.+data OwnershipEvidence = OwnershipMatches | OwnershipMissing | OwnershipMismatch+  deriving stock (Eq, Show)++-- | Initialize a dedicated, absent version-2 schema or transactionally upgrade+-- an owned version-1 catalog. An occupied schema without Hinagata's version+-- marker is refused. The administration role must own an existing schema.+ensureCatalog :: ConnectionTarget -> SqlIdentifier -> SessionOptions -> IO (Either CatalogError CatalogIdentity)+ensureCatalog target schema options = do+  opened <- withSession target options $ \session ->+    runExclusive session $ \connection sessionOptions -> bootstrap connection schema sessionOptions+  pure $ case opened of+    Left failure -> Left (sessionFailure failure)+    Right (Left failure) -> Left (sessionFailure failure)+    Right (Right result) -> result++-- | Validate an existing catalog without creating a schema or changing any+-- catalog row. Cleanup previews use this entry point to remain read-only.+inspectCatalog :: ConnectionTarget -> SqlIdentifier -> SessionOptions -> IO (Either CatalogError CatalogIdentity)+inspectCatalog target schema options = do+  opened <- withSession target options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      let name = Encoding.encodeUtf8 (sqlIdentifierText schema)+      namespace <- queryParamRows connection deadline Sql "SELECT pg_get_userbyid(nspowner), current_user FROM pg_namespace WHERE nspname = $1" [Just name] 1+      checked <- case namespace of+        Left failure -> pure (Left (nativeFailure failure))+        Right [[Just owner, Just current]]+          | owner == current -> do+              inspected <- validateCatalog connection schema deadline+              pure $ case inspected of+                Right CatalogIdentity {formatVersion = 1} -> Left (CatalogError "maintenance catalog requires a version-2 upgrade through ensureCatalog" Nothing Nothing)+                other -> other+          | otherwise -> pure (Left (CatalogError "maintenance schema is owned by another role" Nothing Nothing))+        Right [] -> pure (Left (CatalogError "maintenance catalog does not exist" Nothing Nothing))+        Right _ -> pure (Left (CatalogError "maintenance schema identity is ambiguous" Nothing Nothing))+      pure (Keep checked)+  pure $ case opened of+    Left failure -> Left (sessionFailure failure)+    Right (Left failure) -> Left (sessionFailure failure)+    Right (Right result) -> result++-- | Recheck a recorded name, OID, and marker before any destructive action.+-- A missing or replaced database is reported without changing it.+verifyOwnedDatabase :: ConnectionTarget -> SessionOptions -> CatalogIdentity -> OwnedDatabase -> IO (Either CatalogError OwnershipEvidence)+verifyOwnedDatabase target options identity expected = do+  opened <- withSession target options $ \session ->+    runExclusive session $ \connection sessionOptions -> do+      deadline <- deadlineAfter (operationDeadlineMs sessionOptions)+      checked <- checkOwnership connection deadline identity expected+      pure (either (Retire . Left) (Keep . Right) checked)+  pure $ case opened of+    Left failure -> Left (sessionFailure failure)+    Right (Left failure) -> Left (sessionFailure failure)+    Right (Right result) -> result++checkOwnership :: PQ.Connection -> Deadline -> CatalogIdentity -> OwnedDatabase -> IO (Either CatalogError OwnershipEvidence)+checkOwnership connection deadline identity expected = do+  let statement = "SELECT d.oid::text, s.description, pg_get_userbyid(d.datdba), current_user FROM pg_database d LEFT JOIN pg_shdescription s ON s.objoid = d.oid AND s.classoid = 'pg_database'::regclass WHERE d.datname = $1"+      expectedMarker = "hinagata:v1:" <> clusterUuid identity <> ":" <> token expected+  rows <- queryParamRows connection deadline Sql statement [Just (Encoding.encodeUtf8 (databaseNameText (name expected)))] 1+  pure $ case rows of+    Left failure -> Left (nativeFailure failure)+    Right [] -> Right OwnershipMissing+    Right [[Just actualOid, Just marker, Just owner, Just current]]+      | actualOid == Encoding.encodeUtf8 (Text.pack (show (oid expected))) && marker == Encoding.encodeUtf8 expectedMarker && owner == current -> Right OwnershipMatches+    Right _ -> Right OwnershipMismatch++bootstrap :: PQ.Connection -> SqlIdentifier -> SessionOptions -> IO (SessionDisposition (Either CatalogError CatalogIdentity))+bootstrap connection schema options = do+  deadline <- deadlineAfter (operationDeadlineMs options)+  begun <- query connection deadline Begin "BEGIN"+  case begun of+    Left failure -> pure (Retire (Left (nativeFailure failure)))+    Right () -> do+      result <- initialize connection schema deadline+      case result of+        Left failure -> rollbackCatalog connection options failure+        Right identity -> do+          committed <- query connection deadline Commit "COMMIT"+          case committed of+            Left failure -> pure (Retire (Left (nativeFailure failure)))+            Right () -> do+              status <- PQ.transactionStatus connection+              if status == PQ.TransIdle+                then pure (Keep (Right identity))+                else pure (Retire (Left (CatalogError "catalog transaction did not return to idle" Nothing Nothing)))++initialize :: PQ.Connection -> SqlIdentifier -> Deadline -> IO (Either CatalogError CatalogIdentity)+initialize connection schema deadline = do+  let name = Encoding.encodeUtf8 (sqlIdentifierText schema)+  locked <- queryParamRows connection deadline Sql "SELECT pg_advisory_xact_lock(1212761905, hashtext($1))" [Just name] 1+  case locked of+    Left failure -> pure (Left (nativeFailure failure))+    Right _ -> do+      namespace <- queryParamRows connection deadline Sql "SELECT pg_get_userbyid(nspowner), current_user FROM pg_namespace WHERE nspname = $1" [Just name] 1+      case namespace of+        Left failure -> pure (Left (nativeFailure failure))+        Right [] -> createCatalog connection schema deadline+        Right [[Just owner, Just current]]+          | owner == current -> do+              existing <- validateCatalog connection schema deadline+              case existing of+                Right CatalogIdentity {formatVersion = 1} -> upgradeCatalog connection schema deadline+                other -> pure other+          | otherwise -> pure (Left (CatalogError "maintenance schema is owned by another role" Nothing Nothing))+        Right _ -> pure (Left (CatalogError "maintenance schema identity is ambiguous" Nothing Nothing))++createCatalog :: PQ.Connection -> SqlIdentifier -> Deadline -> IO (Either CatalogError CatalogIdentity)+createCatalog connection schema deadline = do+  file <- getDataFileName "sql/catalog-v2.sql"+  source <- try @IOException (ByteString.readFile file)+  case source of+    Left _ -> pure (Left (CatalogError "catalog DDL is unavailable" Nothing Nothing))+    Right bytes -> do+      let statement =+            Encoding.encodeUtf8 $+              Text.replace "%SCHEMA%" (quoteSqlIdentifier schema) (Encoding.decodeUtf8 bytes)+      created <- query connection deadline Sql statement+      case created of+        Left failure -> pure (Left (nativeFailure failure))+        Right () -> validateCatalog connection schema deadline++upgradeCatalog :: PQ.Connection -> SqlIdentifier -> Deadline -> IO (Either CatalogError CatalogIdentity)+upgradeCatalog connection schema deadline = do+  file <- getDataFileName "sql/upgrade-v1-v2.sql"+  source <- try @IOException (ByteString.readFile file)+  case source of+    Left _ -> pure (Left (CatalogError "catalog upgrade DDL is unavailable" Nothing Nothing))+    Right bytes -> do+      let statement = Encoding.encodeUtf8 (Text.replace "%SCHEMA%" (quoteSqlIdentifier schema) (Encoding.decodeUtf8 bytes))+      upgraded <- query connection deadline Sql statement+      case upgraded of+        Left failure -> pure (Left (nativeFailure failure))+        Right () -> validateCatalog connection schema deadline++validateCatalog :: PQ.Connection -> SqlIdentifier -> Deadline -> IO (Either CatalogError CatalogIdentity)+validateCatalog connection schema deadline = do+  let name = Encoding.encodeUtf8 (sqlIdentifierText schema)+      structureQuery = "SELECT count(*)::text FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = $1 AND c.relname IN ('meta', 'generations', 'allocations', 'leases') AND c.relkind = 'r' AND c.relowner = current_user::regrole"+  structure <- queryParamRows connection deadline Sql structureQuery [Just name] 1+  case structure of+    Left failure -> pure (Left (nativeFailure failure))+    Right [[Just "4"]] -> do+      let statement = Encoding.encodeUtf8 ("SELECT format_version::text, cluster_uuid::text FROM " <> quoteSqlIdentifier schema <> ".\"meta\" WHERE singleton")+      rows <- queryParamRows connection deadline Sql statement [] 2+      case rows of+        Left failure -> pure (Left (nativeFailure failure))+        Right [[Just "1", Just uuid]] -> pure (Right (CatalogIdentity 1 (Encoding.decodeUtf8 uuid)))+        Right [[Just "2", Just uuid]] -> do+          let columnsQuery = "SELECT count(*)::text FROM information_schema.columns WHERE table_schema = $1 AND table_name IN ('generations', 'allocations') AND column_name = 'last_error' AND data_type = 'text'"+          columns <- queryParamRows connection deadline Sql columnsQuery [Just name] 1+          pure $ case columns of+            Left failure -> Left (nativeFailure failure)+            Right [[Just "2"]] -> Right (CatalogIdentity 2 (Encoding.decodeUtf8 uuid))+            Right _ -> Left (CatalogError "maintenance catalog lacks version-2 columns" Nothing Nothing)+        Right _ -> pure (Left (CatalogError "maintenance catalog has an unsupported or invalid format" Nothing Nothing))+    Right _ -> pure (Left (CatalogError "maintenance schema lacks a complete Hinagata catalog" Nothing Nothing))++rollbackCatalog :: PQ.Connection -> SessionOptions -> CatalogError -> IO (SessionDisposition (Either CatalogError CatalogIdentity))+rollbackCatalog connection options failure = do+  deadline <- deadlineAfter (cleanupDeadlineMs options)+  rolled <- query connection deadline Rollback "ROLLBACK"+  case rolled of+    Left cleanup -> pure (Retire (Left failure {cleanupFailure = Just (reason cleanup)}))+    Right () -> do+      status <- PQ.transactionStatus connection+      if status == PQ.TransIdle+        then pure (Keep (Left failure))+        else pure (Retire (Left failure {cleanupFailure = Just "catalog rollback did not return to idle"}))++nativeFailure :: NativeError -> CatalogError+nativeFailure NativeError {reason, sqlState} = CatalogError reason sqlState Nothing++sessionFailure :: SessionError -> CatalogError+sessionFailure failure = CatalogError (Text.pack (show failure)) Nothing Nothing
+ src/Hinagata/Postgres/Session.hs view
@@ -0,0 +1,10 @@+-- | Bracketed, exclusive libpq sessions for Hinagata operations.+module Hinagata.Postgres.Session+  ( Session,+    SessionOptions (..),+    defaultSessionOptions,+    withSession,+  )+where++import Hinagata.Postgres.Internal.Libpq
+ test/Main.hs view
@@ -0,0 +1,1684 @@+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)
+ test/stalled-postgres-peer.py view
@@ -0,0 +1,37 @@+"""Test-only PostgreSQL startup peer: accept a connection, then observe close."""+import pathlib+import socket+import struct+import sys++root = pathlib.Path(sys.argv[1])+with socket.socket() as listener:+    listener.bind(("127.0.0.1", 0))+    listener.listen(1)+    (root / "port").write_text(str(listener.getsockname()[1]))+    with listener.accept()[0] as peer:+        def receive(size):+            data = b""+            while len(data) < size:+                block = peer.recv(size - len(data))+                if not block:+                    return None+                data += block+            return data++        while True:+            header = receive(4)+            if header is None:+                break+            packet = receive(struct.unpack("!I", header)[0] - 4)+            if packet is None:+                break+            # Refuse SSL/GSS negotiation; accept and stall the startup packet.+            if len(packet) == 4:+                peer.sendall(b"N")+                continue+            (root / "accepted").write_text("ready")+            while peer.recv(4096):+                pass+            break+    (root / "closed").write_text("EOF")