diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,47 @@ ## [Unreleased] +## 0.20.0.0 — 2026-10-12 + +### Breaking Changes + +- Adopt `kiroku-store ^>=0.11.0.0`, migrations `^>=0.7.1.0` and the + Shibuya adapter `^>=0.6.0.1`. Apply Kiroku migrations 0013 and 0014 before + starting this cohort. Subscription sizes, members, batches and buffers use + validated public constructors. `ShardWorkerError` gains `ShardReaderStartupFailed`; + `ShardCountMismatch` can be caught as `SomeSubscriptionStartupFailure`. +- The inspection cohort from mori://shinzui/kiroku/packages/kiroku-store + adds five constructors to the closed `Store` effect; exhaustive custom + interpreters must handle them. Complete `Subscriber` record construction + must supply `subDropped`. The composed plan includes migration 0015 from + mori://shinzui/kiroku/packages/kiroku-store-migrations, adding the stream + browsing index. Its transactional index build can block writes; schedule + a write pause for large persistent stores. + +### New Features + +- Add `runDomainCommandWithCatalogProjectionsGuarded`, an opt-in relational + precondition after catalog admission and tentative append. Typed refusal rolls + back the batch; accepted delivery uses the same protected serving bindings. + +- Add `resizeShardCount` and transaction-composable `resizeShardCountTx` for + stopped, unowned groups. Atomically rewind every checkpoint to the previous + minimum and replace lease rows, with typed owner refusal and resize evidence. + +### Bug Fixes + +- Refuse mismatched shard counts before inserting lease rows. Classify typed + event-decode failures as deterministic failures, preserving stop-on-undecodable + subscription behavior. + +### Other Changes + +- Require [`shibuya-core`](mori://shinzui/shibuya/packages/shibuya-core) + `^>=0.10.0.1`, including its random 1.3 compatibility. Existing processor + APIs are unchanged. +- Clarify asynchronous query models as the default, minimal synchronous command + authority, and configurable bounded router/process-manager retries. + ## 0.19.0.1 — 2026-10-05 ### Other Changes diff --git a/bench/GuardedCatalogBaselineMain.hs b/bench/GuardedCatalogBaselineMain.hs new file mode 100644 --- /dev/null +++ b/bench/GuardedCatalogBaselineMain.hs @@ -0,0 +1,24 @@ +{-# LANGUAGE BlockArguments #-} + +module Main (main) where + +import Data.Aeson qualified as Aeson +import Data.ByteString.Lazy qualified as Lazy +import GuardedCatalogWorkload +import Keiro.Test.Postgres +import System.Environment (getArgs) +import Prelude + +-- Build this same harness against the recorded original library source. The +-- parent invokes it once per paired sample on an equivalent fresh database. +main :: IO () +main = do + arguments <- getArgs + case arguments of + [input, output] -> do + config <- Lazy.readFile input >>= either fail pure . Aeson.eitherDecode + withMigratedBenchmarkSuite \fixture -> withFreshResourceStore fixture \(_, runner) -> do + prepareDatabase runner config + result <- measureWorkload runner config (ordinaryAction runner (batchSize config)) + Lazy.writeFile output (Aeson.encode result) + _ -> fail "usage: ordinary-baseline WORKLOAD.json RESULT.json" diff --git a/bench/GuardedCatalogBench.hs b/bench/GuardedCatalogBench.hs new file mode 100644 --- /dev/null +++ b/bench/GuardedCatalogBench.hs @@ -0,0 +1,386 @@ +{-# LANGUAGE BlockArguments #-} + +module GuardedCatalogBench (runGuardedCatalogEvidence) where + +import Control.Concurrent +import Control.Exception (AsyncException (ThreadKilled), finally, fromException) +import Control.Monad +import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KeyMap +import Data.ByteString.Lazy qualified as Lazy +import Data.IORef +import Data.List qualified as List +import Data.Text qualified as Text +import GuardedCatalogExample qualified as Example +import GuardedCatalogWorkload +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Statement (preparable) +import Keiro +import Keiro.Prelude +import Keiro.Projection +import Keiro.Test.Postgres +import Kiroku.Store qualified as Store +import System.Directory +import System.Environment (getArgs, getExecutablePath, lookupEnv) +import System.Exit (ExitCode (..)) +import System.Process (readProcessWithExitCode) +import "hasql-transaction" Hasql.Transaction qualified as Tx +import Prelude + +data Mode = Ordinary | EmptyGuard | IndexedGuard | SlowGuard deriving (Eq, Show) + +runGuardedCatalogEvidence :: IO Bool +runGuardedCatalogEvidence = + lookupEnv "KEIRO_GUARDED_CATALOG_EVIDENCE" >>= \case + Nothing -> pure False + Just "single-ordinary" -> do + getArgs >>= \case + [input, output] -> do + config <- Lazy.readFile input >>= either fail pure . Aeson.eitherDecode + result <- sample config Ordinary + Lazy.writeFile output (Aeson.encode result) + _ -> fail "single-ordinary requires workload and result paths" + pure True + Just mode -> do + output <- maybe "bench/results/guarded-catalog/candidate" id <$> lookupEnv "KEIRO_GUARDED_CATALOG_OUTPUT" + exists <- doesPathExist (output <> "/manifest.json") + when exists (fail "evidence output already exists; choose a unique output directory") + createDirectoryIfMissing True output + -- Fixed budgets are persisted before any candidate sample. + Lazy.writeFile + (output <> "/manifest.json") + ( Aeson.encode + ( Aeson.object + [ "mode" Aeson..= mode, + "pairs" Aeson..= (5 :: Int), + "attempts" Aeson..= (10000 :: Int), + "payloadBytes" Aeson..= (1024 :: Int), + "tags" Aeson..= (4 :: Int), + "poolSize" Aeson..= (10 :: Int), + "ordinaryThroughputMinimum" Aeson..= (0.95 :: Double), + "ordinaryP95Maximum" Aeson..= (1.10 :: Double), + "ordinaryP99Maximum" Aeson..= (1.10 :: Double), + "guardedThroughputMinimum" Aeson..= (0.90 :: Double), + "guardedP95Maximum" Aeson..= (1.25 :: Double), + "guardedP99Maximum" Aeson..= (1.50 :: Double), + "requestSeed" Aeson..= (306 :: Int), + "deadlineSeconds" Aeson..= (300 :: Int) + ] + ) + ) + case mode of + "ordinary" -> forM_ ordinaryConfigs \config -> do + result <- sample config Ordinary + save output (workloadKey config <> "-ordinary") result + "compare" -> do + baseline <- requireBaseline + baselineFolder <- lookupEnv "KEIRO_GUARDED_CATALOG_BASELINE" >>= maybe (fail "baseline directory required") pure + copyFile (baselineFolder <> "/enhanced-environment.json") (output <> "/baseline-environment.json") + -- Stop on a failed fixed budget; retain every sample and report. This + -- prevents continuing large measurements on a known unacceptable path. + forM_ ordinaryConfigs \config -> comparePairSet output baseline "refactor" Ordinary config + forM_ ordinaryConfigs \base -> do + comparePairSet output baseline "empty" EmptyGuard (base {guardPercent = 100}) + comparePairSet output baseline "miss" IndexedGuard (base {guardPercent = 100}) + forM_ [False, True] \spread -> do + -- Pure-hit pairs report costs, without a numeric acceptance + -- budget: the candidate rolls back every creation batch. + comparePairSet output baseline "hit" IndexedGuard (base {guardPercent = 100, duplicatePercent = 100, distributed = spread}) + forM_ [0, 50, 100] \duplicates -> comparePairSet output baseline "mixed" IndexedGuard (base {guardPercent = 20, duplicatePercent = duplicates, distributed = spread}) + -- Offered rates are fixed from controls before these candidate runs. + forM_ [False, True] \worker -> do + let base = Workload 1000000 8 4 worker 20 0 True Nothing 10000 300 + capacities <- forM [1 .. 5 :: Int] \iteration -> do + capacity <- sample base Ordinary + save output ("capacity-" <> show worker <> "-" <> show iteration) capacity + pure (fromIntegral (length (samples capacity)) / elapsedSeconds capacity) + let rate = percentile 0.5 capacities + Lazy.writeFile (output <> "/capacity-" <> show worker <> "-rates.json") (Aeson.encode (Aeson.object ["samples" Aeson..= capacities, "fixedCapacity" Aeson..= rate])) + forM_ [0.50, 0.80] \fraction -> forM_ [0, 50, 100] \duplicates -> + comparePairSet output baseline "fixed-arrival" IndexedGuard (base {duplicatePercent = duplicates, arrivalRate = Just (rate * fraction), deadlineSeconds = max 300 (ceiling (10000 / (rate * fraction)) + 60)}) + -- Paired ten-minute runs preserve normal vacuum and async backlog. + let sustained = base {duplicatePercent = 100, arrivalRate = Just (rate * 0.8), attempts = max 10000 (ceiling (rate * 0.8 * 600) + 1), deadlineSeconds = 720} + sustainedResults <- forM [("control", Ordinary, sustained {duplicatePercent = 0}), ("duplicates", IndexedGuard, sustained)] \(label, chosen, config) -> do + result <- sampleWithStorage output ("sustained-" <> show worker <> "-" <> label) config chosen + save output ("sustained-" <> show worker <> "-" <> label) result + pure result + case sustainedResults of + [control, candidate] -> do + let comparison = ratios "mixed" control candidate + minuteRatios = [ratios "mixed" (minute n control) (minute n candidate) | n <- [0 .. 9 :: Int]] + success = passes "mixed" [comparison] && all (passes "mixed" . (: [])) minuteRatios && all ((>= 600) . elapsedSeconds) sustainedResults + Lazy.writeFile (output <> "/sustained-" <> show worker <> "-report.json") (Aeson.encode (Aeson.object ["latencyPass" Aeson..= success, "overall" Aeson..= comparison, "minutes" Aeson..= minuteRatios, "storageReviewRequired" Aeson..= True])) + unless success (fail "sustained ordinary-writer latency budget failed") + _ -> fail "missing sustained pair" + "diagnostic" -> do + let config = Workload 1000000 8 4 False 20 0 True Nothing 1000 300 + control <- sample config Ordinary + slow <- sample config SlowGuard + save output "slow-control" control + save output "slow-guard" slow + unless (not (passes "mixed" [ratios "mixed" control slow])) (fail "slow guard failed to demonstrate ordinary-writer degradation") + withMigratedBenchmarkSuite \fixture -> withFreshResourceStore fixture \(_, runner) -> do + prepareDatabase runner config + plan <- + checked + runner + ( Store.runTransaction + ( Tx.statement + (Example.notePayload (Example.exampleNote "explain" "seed-1")) + ( preparable + "EXPLAIN (ANALYZE, BUFFERS) SELECT note_id FROM app.state_note_identity WHERE anchor_type=$1::jsonb->>'anchorType' AND anchor_id=$1::jsonb->>'anchorId' AND property_id=$1::jsonb->>'propertyId' AND state_id=$1::jsonb->>'stateId'" + Example.textParam + (D.rowList (D.column (D.nonNullable D.text))) + ) + ) + ) + writeFile (output <> "/identity-query-plan.txt") (unlines (map Text.unpack plan)) + unless (any (Text.isInfixOf "Index") plan) (fail "identity lookup did not use an index") + diagnosticTransactions output config + _ -> fail ("unknown evidence mode: " <> mode) + pure True + +ordinaryConfigs :: [Workload] +ordinaryConfigs = [Workload size connections batch worker 0 0 True Nothing 10000 300 | size <- [10000, 1000000], connections <- [1, 8], batch <- [1, 4], worker <- [False, True]] + +workloadKey :: Workload -> String +workloadKey config = List.intercalate "-" [show (cardinality config), show (connections config), show (batchSize config), show (activeWorker config), show (guardPercent config), show (duplicatePercent config), show (distributed config), show (arrivalRate config)] + +sample :: Workload -> Mode -> IO Measurements +sample config mode = withMigratedBenchmarkSuite \fixture -> withFreshResourceStore fixture \(_, runner) -> do + prepareDatabase runner config + measureWorkload runner config (action runner config mode) + +action :: StoreRunner -> Workload -> Mode -> Request -> IO Bool +action runner config mode request + | mode == Ordinary || not (guarded request) = ordinaryAction runner (batchSize config) request + | otherwise = do + outcome <- + checked + runner + ( runDomainCommandWithCatalogProjectionsGuarded + defaultRunCommandOptions + (Example.handler (batchSize config)) + (stream (Example.noteId (note request))) + (Example.CreateNote (payload request)) + Example.catalog + Example.commandSet + ( \context -> do + when (mode == SlowGuard) (Tx.sql "SELECT pg_sleep(0.02)") + if mode == EmptyGuard || mode == SlowGuard then pure (Right ()) else Example.noteGuard (note request) context + ) + ) + case outcome of + Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}})) -> pure True + Right (PreconditionRefused (Example.ExistingNote _)) -> pure False + other -> fail ("unexpected guarded outcome: " <> show other) + +save :: FilePath -> String -> Measurements -> IO () +save output name result = do + let raw = output <> "/" <> name <> ".json" + Lazy.writeFile raw (Aeson.encode result) + (code, _, errors) <- readProcessWithExitCode "gzip" ["-n", raw] "" + unless (code == ExitSuccess) (fail ("raw evidence compression failed: " <> errors)) + Lazy.writeFile (output <> "/" <> name <> "-summary.json") (Aeson.encode (summarize result)) + putStrLn (name <> " " <> show (summarize result)) + +requireBaseline :: IO FilePath +requireBaseline = do + folder <- lookupEnv "KEIRO_GUARDED_CATALOG_BASELINE" >>= maybe (fail "baseline directory required") pure + manifest <- Lazy.readFile (folder <> "/enhanced-environment.json") >>= either fail pure . Aeson.eitherDecode + (path, digest, workloadDigest, fixtureDigest) <- case manifest of + Aeson.Object object -> case (KeyMap.lookup "retainedExecutable" object, KeyMap.lookup "sha256" object, KeyMap.lookup "workloadSha256" object, KeyMap.lookup "fixtureSha256" object) of + (Just (Aeson.String path), Just (Aeson.String digest), Just (Aeson.String workloadDigest), Just (Aeson.String fixtureDigest)) -> pure (Text.unpack path, Text.unpack digest, Text.unpack workloadDigest, Text.unpack fixtureDigest) + _ -> fail "baseline manifest is incomplete" + _ -> fail "invalid baseline manifest" + forM_ [(path, digest), ("bench/GuardedCatalogWorkload.hs", workloadDigest), ("test/GuardedCatalogFixture.hs", fixtureDigest)] \(file, expected) -> do + (code, actual, errors) <- readProcessWithExitCode "shasum" ["-a", "256", file] "" + unless (code == ExitSuccess && take 64 actual == expected) (fail ("baseline/build configuration digest mismatch: " <> file <> errors)) + pure path + +comparePairSet :: FilePath -> FilePath -> String -> Mode -> Workload -> IO () +comparePairSet output baseline label mode config = do + pairs <- forM [1 .. 5 :: Int] \pair -> do + let prefix = label <> "-" <> workloadKey config <> "-" <> show pair + controlConfig = config {duplicatePercent = 0} + control = if label == "refactor" then originalSample output baseline prefix controlConfig else sample controlConfig Ordinary + -- Both old and new ordinary binaries get a fresh RTS process. A + -- long-lived parent otherwise has different GC history from each + -- newly launched retained control, despite identical request code. + candidate = if label == "refactor" then currentOrdinarySample output prefix config else sample config mode + (before, after) <- if odd pair then (,) <$> control <*> candidate else do after <- candidate; before <- control; pure (before, after) + save output (prefix <> "-control") before + save output (prefix <> "-candidate") after + pure (ratios label before after) + let accepted = label == "hit" || passes label pairs + Lazy.writeFile (output <> "/" <> label <> "-" <> workloadKey config <> "-report.json") (Aeson.encode (Aeson.object ["pass" Aeson..= accepted, "budgetApplicable" Aeson..= (label /= "hit"), "ratios" Aeson..= pairs])) + unless accepted (fail ("fixed performance budget failed: " <> label <> " " <> workloadKey config <> " " <> show pairs)) + +originalSample output executable prefix config = do + let input = output <> "/" <> prefix <> "-workload.json"; result = output <> "/" <> prefix <> "-original.json" + Lazy.writeFile input (Aeson.encode config) + (exit, stdout, stderr) <- readProcessWithExitCode executable [input, result, "+RTS", "-N2", "-RTS"] "" + unless (exit == ExitSuccess) (fail ("original benchmark fault: " <> stdout <> stderr)) + measurements <- Lazy.readFile result >>= either fail pure . Aeson.eitherDecode + (code, _, errors) <- readProcessWithExitCode "gzip" ["-n", result] "" + unless (code == ExitSuccess) (fail ("baseline evidence compression failed: " <> errors)) + pure measurements + +currentOrdinarySample :: FilePath -> String -> Workload -> IO Measurements +currentOrdinarySample output prefix config = do + executable <- getExecutablePath + let input = output <> "/" <> prefix <> "-candidate-workload.json" + result = output <> "/" <> prefix <> "-candidate-process.json" + Lazy.writeFile input (Aeson.encode config) + (exit, stdout, stderr) <- readProcessWithExitCode "env" ["KEIRO_GUARDED_CATALOG_EVIDENCE=single-ordinary", executable, input, result, "+RTS", "-N2", "-RTS"] "" + unless (exit == ExitSuccess) (fail ("candidate subprocess fault: " <> stdout <> stderr)) + measurements <- Lazy.readFile result >>= either fail pure . Aeson.eitherDecode + (compressed, _, compressionErrors) <- readProcessWithExitCode "gzip" ["-n", result] "" + unless (compressed == ExitSuccess) (fail ("candidate compression fault: " <> compressionErrors)) + pure measurements + +-- Ratios retain throughput and the relevant end-to-end latency cohort. Refused +-- and accepted counts are separately reported in each sample summary. +ratios :: String -> Measurements -> Measurements -> (Double, Double, Double) +ratios label before after = (throughput after / throughput before, percentile 0.95 (latencies after) / percentile 0.95 (latencies before), percentile 0.99 (latencies after) / percentile 0.99 (latencies before)) + where + throughput value = fromIntegral (length (samples value)) / elapsedSeconds value + latencies value = [latencyMicros sample | sample <- samples value, if label == "mixed" || label == "fixed-arrival" then cohort sample == "ordinary" else True] + +passes :: String -> [(Double, Double, Double)] -> Bool +passes label values = median [x | (x, _, _) <- values] >= (if label == "refactor" then 0.95 else 0.90) && median [x | (_, x, _) <- values] <= (if label == "refactor" then 1.10 else 1.25) && median [x | (_, _, x) <- values] <= (if label == "refactor" then 1.10 else 1.50) + where + median = percentile 0.5 + +sampleWithStorage output name config mode = withMigratedBenchmarkSuite \fixture -> withFreshResourceStore fixture \(_, runner) -> do + prepareDatabase runner config + let path = output <> "/" <> name <> "-storage.jsonl" + snapshot = do + snapshot <- + checked + runner + ( Store.runTransaction + ( Tx.statement + () + ( preparable + "SELECT jsonb_build_object('at',clock_timestamp(),'wal',(SELECT row_to_json(w) FROM pg_stat_wal w),'databaseBytes',pg_database_size(current_database()),'tables',(SELECT jsonb_agg(jsonb_build_object('schema',schemaname,'table',relname,'bytes',pg_total_relation_size(relid),'relationBytes',pg_relation_size(relid),'indexBytes',pg_indexes_size(relid),'live',n_live_tup,'dead',n_dead_tup,'autovacuumCount',autovacuum_count,'lastAutovacuum',last_autovacuum)) FROM pg_stat_user_tables),'head',(SELECT stream_version FROM streams WHERE stream_id=0),'worker',(SELECT min(last_seen) FROM subscriptions WHERE subscription_name='note-rich-worker'))::text" + E.noParams + (D.singleRow (D.column (D.nonNullable D.text))) + ) + ) + ) + appendFile path (Text.unpack snapshot <> "\n") + collect = do + threadDelay (60 * 1000000) + snapshot + collect + snapshot + done <- newEmptyMVar + observer <- forkFinally collect (putMVar done) + result <- measureWorkload runner config (action runner config mode) `finally` killThread observer + takeMVar done >>= \case + Left err -> case fromException err of + Just ThreadKilled -> pure () + _ -> fail ("storage observer failed: " <> show err) + Right () -> fail "storage observer exited unexpectedly" + snapshot + pure result + +minute :: Int -> Measurements -> Measurements +minute n result = result {elapsedSeconds = 60, samples = [sample | sample <- samples result, completedSeconds sample >= fromIntegral (n * 60), completedSeconds sample < fromIntegral ((n + 1) * 60)]} + +-- These extra statements/triggers run only in diagnostic mode. The clock +-- interval starts after the global row update and ends before commit, so it is +-- a lower bound on that row's lock lifetime, not a production hold-time sample. +diagnosticTransactions :: FilePath -> Workload -> IO () +diagnosticTransactions output config = withMigratedBenchmarkSuite \fixture -> + withFreshResourceStoresSameDatabase fixture \((_, runner), (_, observerRunner)) -> do + prepareDatabase runner config + checked + runner + ( Store.runTransaction + ( Tx.sql + """ + CREATE SEQUENCE app.diag_all_attempts; + CREATE SEQUENCE app.diag_guard_attempts; + CREATE TABLE app.diag_marks(txid bigint, stage text, at timestamptz); + CREATE FUNCTION app.diag_all() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + PERFORM nextval('app.diag_all_attempts'); + INSERT INTO app.diag_marks VALUES (txid_current(), 'all-updated', clock_timestamp()); + RETURN NEW; + END $$; + CREATE TRIGGER diag_all AFTER UPDATE ON streams FOR EACH ROW WHEN (NEW.stream_id = 0) EXECUTE FUNCTION app.diag_all(); + CREATE FUNCTION app.diag_identity() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO app.diag_marks VALUES (txid_current(), 'identity-delivered', clock_timestamp()); + RETURN NEW; + END $$; + CREATE TRIGGER diag_identity AFTER INSERT ON app.state_note_identity FOR EACH ROW EXECUTE FUNCTION app.diag_identity(); + CREATE FUNCTION app.diag_append_end() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO app.diag_marks VALUES (txid_current(), 'append-statement-end', clock_timestamp()); + RETURN NULL; + END $$; + CREATE TRIGGER diag_append_end AFTER INSERT ON stream_events FOR EACH STATEMENT EXECUTE FUNCTION app.diag_append_end(); + """ + ) + ) + occ <- newIORef (0 :: Int) + done <- newEmptyMVar + let observe = do + waits <- checked observerRunner (Store.runTransaction (Tx.statement () (preparable "SELECT COALESCE(jsonb_agg(jsonb_build_object('at',clock_timestamp(),'pid',pid,'waitType',wait_event_type,'waitEvent',wait_event,'blockers',pg_blocking_pids(pid),'query',query)),'[]'::jsonb)::text FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock'" E.noParams (D.singleRow (D.column (D.nonNullable D.text)))))) + appendFile (output <> "/lock-waits.jsonl") (Text.unpack waits <> "\n") + threadDelay 20000 + observe + options = defaultRunCommandOptions & #beforeAppend .~ atomicModifyIORef' occ (\n -> (n + 1, ())) + execute request = do + let target = stream (Example.noteId (note request)); command = Example.CreateNote (payload request) + if guarded request + then do + result <- + checked + runner + ( runDomainCommandWithCatalogProjectionsGuarded + options + (Example.handler (batchSize config)) + target + command + Example.catalog + Example.commandSet + ( \context -> do + attempt <- Tx.statement () (preparable "SELECT nextval('app.diag_guard_attempts')" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))) + -- One known driver retry makes surviving attempt counters a + -- positive control instead of assuming each body ran once. + when (attempt == 1) (Tx.sql "DO $$ BEGIN RAISE EXCEPTION 'diagnostic retry' USING ERRCODE='40001'; END $$") + Tx.sql "INSERT INTO app.diag_marks VALUES (txid_current(),'guard-start',clock_timestamp())" + Tx.sql "SELECT pg_sleep(0.02)" + refusal <- Example.noteGuard (note request) context + Tx.sql "INSERT INTO app.diag_marks VALUES (txid_current(),'guard-end',clock_timestamp())" + pure refusal + ) + ) + case result of + Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}})) -> pure True + other -> fail ("diagnostic guarded fault: " <> show other) + else do + result <- checked runner (runDomainCommandWithCatalogProjections options (Example.handler (batchSize config)) target command Example.catalog Example.commandSet) + case result of + Right (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}}) -> pure True + other -> fail ("diagnostic ordinary fault: " <> show other) + observer <- forkFinally observe (putMVar done) + result <- measureWorkload runner config execute `finally` killThread observer + takeMVar done >>= \case + Left err -> case fromException err of Just ThreadKilled -> pure (); _ -> fail ("lock observer failed: " <> show err) + Right () -> fail "lock observer exited unexpectedly" + save output "instrumented-slow-guard" result + attemptsObserved <- readIORef occ + (allUpdates, guardEvaluations) <- checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT (SELECT last_value FROM app.diag_all_attempts), (SELECT last_value FROM app.diag_guard_attempts)" E.noParams (D.singleRow ((,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8)))))) + unless (allUpdates == fromIntegral (attempts config + 1) && guardEvaluations == fromIntegral (length (filter guarded (prepareRequests config)) + 1)) (fail "diagnostic driver retry count differs from injected retry") + waits <- lines <$> readFile (output <> "/lock-waits.jsonl") + unless (any (/= "[]") waits) (fail "slow-guard diagnostic observed no lock wait") + statistics <- checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT jsonb_build_object('appendAllUpdates',(SELECT last_value FROM app.diag_all_attempts),'guardEvaluations',(SELECT last_value FROM app.diag_guard_attempts),'timings',(SELECT jsonb_agg(row_to_json(t)) FROM (SELECT txid, EXTRACT(EPOCH FROM (max(at) FILTER (WHERE stage='guard-end')-max(at) FILTER (WHERE stage='guard-start')))*1000000 AS guard_microseconds, EXTRACT(EPOCH FROM (max(at) FILTER (WHERE stage='identity-delivered')-max(at) FILTER (WHERE stage='append-statement-end')))*1000000 AS post_append_delivery_microseconds, EXTRACT(EPOCH FROM (max(at) FILTER (WHERE stage='identity-delivered')-min(at) FILTER (WHERE stage='all-updated')))*1000000 AS all_lock_lower_bound_microseconds FROM app.diag_marks GROUP BY txid) t),'rawStages',(SELECT jsonb_agg(row_to_json(m)) FROM app.diag_marks m))::text" E.noParams (D.singleRow (D.column (D.nonNullable D.text)))))) + writeFile (output <> "/transaction-timings.json") (Text.unpack statistics) + Lazy.writeFile (output <> "/attempt-counts.json") (Aeson.encode (Aeson.object ["outerAppendAttempts" Aeson..= attemptsObserved, "commands" Aeson..= attempts config, "injectedDriverRetries" Aeson..= (1 :: Int), "scope" Aeson..= ("diagnostic unique streams; global update/guard sequences count attempts surviving rollback; lock clocks exclude pre-update wait and commit" :: Text)])) + unless (attemptsObserved == attempts config) (fail "unexpected OCC churn in unique-stream diagnostic") diff --git a/bench/GuardedCatalogWorkload.hs b/bench/GuardedCatalogWorkload.hs new file mode 100644 --- /dev/null +++ b/bench/GuardedCatalogWorkload.hs @@ -0,0 +1,242 @@ +{-# LANGUAGE BlockArguments #-} + +-- | Identical request preparation, scheduling and measurement for the retained +-- ordinary binary and candidate. No guarded API dependency lives here. +module GuardedCatalogWorkload where + +import Control.Concurrent +import Control.Exception (AsyncException (ThreadKilled), finally, fromException) +import Control.Monad +import Data.Aeson qualified as Aeson +import Data.Aeson.Key qualified as AesonKey +import Data.ByteString.Lazy qualified as Lazy +import Data.List qualified as List +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Encoding +import Data.Vector qualified as Vector +import Effectful (Eff, IOE) +import Effectful.Error.Static (Error) +import GHC.Clock (getMonotonicTimeNSec) +import GuardedCatalogFixture qualified as Fixture +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Statement (preparable) +import Keiro +import Keiro.Prelude +import Keiro.Projection +import Keiro.Test.Postgres +import Kiroku.Store qualified as Store +import Kiroku.Store.Subscription.Types (MissingCheckpointPolicy (..)) +import Kiroku.Store.Types (GlobalPosition (..)) +import System.Timeout (timeout) +import "hasql-transaction" Hasql.Transaction qualified as Tx +import Prelude + +data Workload = Workload + { cardinality :: Int, + connections :: Int, + batchSize :: Int, + activeWorker :: Bool, + guardPercent :: Int, + duplicatePercent :: Int, + distributed :: Bool, + arrivalRate :: Maybe Double, + attempts :: Int, + deadlineSeconds :: Int + } + deriving stock (Eq, Show, Generic) + deriving anyclass (Aeson.ToJSON, Aeson.FromJSON) + +data Request = Request + { requestNumber :: Int, + guarded :: Bool, + duplicate :: Bool, + note :: Fixture.Note, + payload :: Text + } + +data Sample = Sample + { number :: Int, + cohort :: Text, + accepted :: Bool, + latencyMicros :: Double, + occAttempts :: Maybe Int, + completedSeconds :: Double + } + deriving stock (Eq, Show, Generic) + deriving anyclass (Aeson.ToJSON, Aeson.FromJSON) + +data Measurements = Measurements + { workload :: Workload, + elapsedSeconds :: Double, + samples :: [Sample], + endingCardinality :: Int64, + committedEvents :: Int64, + richRows :: Int64, + headPosition :: Int64, + workerPosition :: Int64, + databaseSettings :: Aeson.Value + } + deriving stock (Show, Generic) + deriving anyclass (Aeson.ToJSON, Aeson.FromJSON) + +prepareRequests :: Workload -> [Request] +prepareRequests config = [request n | n <- [0 .. attempts config - 1]] + where + request n = + let isGuard = n `mod` 100 < guardPercent config + guardOrdinal = (n `div` 100) * guardPercent config + n `mod` 100 + isDuplicate = isGuard && guardOrdinal `mod` 100 < duplicatePercent config + candidate = "measured-note-" <> Text.pack (show n) + key = if isDuplicate then "seed-" <> Text.pack (show (if distributed config then 1 + (n * 104729 + 306) `mod` (cardinality config - 100) else 1)) else candidate + emptyNote = (Fixture.exampleNote candidate key) {Fixture.content = ""} + overhead = fromIntegral (Lazy.length (Aeson.encode (Aeson.String (Fixture.notePayload emptyNote)))) + note = emptyNote {Fixture.content = Text.replicate (1024 - overhead) "x"} + in Request n isGuard isDuplicate note (Fixture.notePayload note) + +prepareDatabase :: StoreRunner -> Workload -> IO () +prepareDatabase runner config = do + durable <- checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT current_setting('fsync')='on' AND current_setting('synchronous_commit')='on' AND current_setting('full_page_writes')='on' AND current_setting('autovacuum')='on'" E.noParams (D.singleRow (D.column (D.nonNullable D.bool)))))) + unless durable (fail "benchmark requires fsync, synchronous_commit, full_page_writes and autovacuum enabled") + Fixture.setup runner + checked + runner + ( Store.runTransaction + ( Tx.sql + ( Encoding.encodeUtf8 + ("INSERT INTO app.state_note_identity SELECT 'intention','seed-'||n,'status','ready','seed-note-'||n FROM generate_series(2," <> Text.pack (show (cardinality config - 99)) <> ") n; ANALYZE app.state_note_identity") + ) + ) + ) + _ <- checked runner (Store.initializeSubscriptionCheckpoint (Store.SubscriptionName "note-rich-worker") 0 FromBeginning) + -- Exactly 100 accepted ordinary warm-up rows are included in initial size. + forM_ [1 .. 100 :: Int] \n -> do + let candidate = "warm-" <> Text.pack (show n) + -- The repeated shared identity has an explicit committed setup + -- creator; distributed hits use the declared bootstrap identities. + note = Fixture.exampleNote candidate (if n == 1 then "seed-1" else candidate) + ordinaryAction runner (batchSize config) (Request n False False note (Fixture.notePayload note)) >>= \accepted -> unless accepted (fail "warm-up refused") + checked runner (Store.runTransaction (Tx.sql "ANALYZE app.state_note_identity")) + when (activeWorker config) (drainWarmup runner (GlobalPosition 0)) + +ordinaryAction :: StoreRunner -> Int -> Request -> IO Bool +ordinaryAction runner batch request = do + outcome <- checked runner (runDomainCommandWithCatalogProjections defaultRunCommandOptions (Fixture.handler batch) (stream (Fixture.noteId (note request))) (Fixture.CreateNote (payload request)) Fixture.catalog Fixture.commandSet) + case outcome of + Right (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}}) -> pure True + other -> fail ("unexpected ordinary outcome: " <> show other) + +measureWorkload :: StoreRunner -> Workload -> (Request -> IO Bool) -> IO Measurements +measureWorkload runner config action = do + let requests = prepareRequests config + -- Force all payload preparation outside timing. + _ <- evaluateRequests requests + queue <- newMVar requests + completions <- replicateM (connections config) newEmptyMVar + asyncDone <- newEmptyMVar + (_, _, _, _, initialCursor) <- facts runner + worker <- if activeWorker config then Just <$> forkFinally (drainWorker runner (GlobalPosition initialCursor)) (putMVar asyncDone) else pure Nothing + start <- getMonotonicTimeNSec + threads <- forM completions \done -> forkFinally (commandWorker start queue []) (putMVar done) + result <- + timeout (deadlineSeconds config * 1000000) (mapM takeMVar completions) + `finally` mapM_ killThread threads + end <- getMonotonicTimeNSec + forM_ worker killThread + forM_ worker \_ -> do + outcome <- takeMVar asyncDone + case outcome of + Left err -> case fromException err of + Just ThreadKilled -> pure () + _ -> fail ("async worker failed: " <> show err) + Right () -> fail "async worker exited unexpectedly" + values <- case result of Nothing -> fail "scheduled work exceeded workload deadline"; Just outcomes -> concat <$> mapM (either (fail . show) pure) outcomes + unless (length values == attempts config) (fail "missing workload attempts") + (identityRows, events, richRows, head, cursor) <- facts runner + settings <- checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT jsonb_object_agg(name, setting)::text FROM pg_settings WHERE name IN ('server_version','fsync','synchronous_commit','full_page_writes','autovacuum','shared_buffers','wal_level','checkpoint_timeout','max_wal_size')" E.noParams (D.singleRow (D.column (D.nonNullable D.text)))))) + databaseSettings <- either fail pure (Aeson.eitherDecodeStrict' (Encoding.encodeUtf8 settings)) + pure (Measurements config (fromIntegral (end - start) / 1e9) values identityRows events richRows head cursor databaseSettings) + where + commandWorker start queue samples = do + request <- modifyMVar queue \case [] -> pure ([], Nothing); x : xs -> pure (xs, Just x) + case request of + Nothing -> pure samples + Just request -> do + let scheduled = case arrivalRate config of + Nothing -> Nothing + Just rate -> Just (start + floor (fromIntegral (requestNumber request) / rate * 1e9)) + forM_ scheduled \arrival -> do + now <- getMonotonicTimeNSec + when (arrival > now) (threadDelay (fromIntegral ((arrival - now) `div` 1000))) + submitted <- getMonotonicTimeNSec + accepted <- action request + unless (accepted == not (duplicate request)) (fail "unexpected accepted/refused schedule outcome") + completed <- getMonotonicTimeNSec + let origin = fromMaybe submitted scheduled + cohort = if guarded request then if accepted then "guarded-accepted" else "guarded-refused" else "ordinary" + commandWorker start queue (Sample (requestNumber request) cohort accepted (fromIntegral (completed - origin) / 1000) Nothing (fromIntegral (completed - start) / 1e9) : samples) + +evaluateRequests :: [Request] -> IO Int +evaluateRequests requests = do + forM_ requests \request -> unless (Lazy.length (Aeson.encode (Aeson.String (payload request))) == 1024 && length (Fixture.tags (note request)) == 4) (fail "creation payload is not exactly 1 KiB with four tags") + pure $! sum [Text.length (payload request) | request <- requests] + +checked :: StoreRunner -> Eff '[Store.Store, Error Store.StoreError, Store.KirokuStoreResource, IOE] a -> IO a +checked (StoreRunner run) action = run action >>= either (fail . show) pure + +drainWorker :: StoreRunner -> GlobalPosition -> IO () +drainWorker runner cursor = do + events <- checked runner (Store.readAllForward cursor 500) + if Vector.null events + then threadDelay 1000 >> drainWorker runner cursor + else do + forM_ events \recorded -> do + outcome <- checked runner (Store.runTransaction (applyAsyncProjectionFromCatalog Fixture.catalog Fixture.richOwner Fixture.richProjection recorded)) + unless (outcome == CatalogAsyncApplied || outcome == CatalogAsyncDuplicate) (fail ("async delivery fault: " <> show outcome)) + let next = Vector.last events ^. #globalPosition + checked runner (Store.runTransaction (Store.resetSubscriptionCheckpointsTx (Store.SubscriptionName "note-rich-worker" :| []) next)) + drainWorker runner next + +drainWarmup :: StoreRunner -> GlobalPosition -> IO () +drainWarmup runner cursor = do + events <- checked runner (Store.readAllForward cursor 500) + unless (Vector.null events) do + forM_ events \recorded -> do + outcome <- checked runner (Store.runTransaction (applyAsyncProjectionFromCatalog Fixture.catalog Fixture.richOwner Fixture.richProjection recorded)) + unless (outcome == CatalogAsyncApplied || outcome == CatalogAsyncDuplicate) (fail ("warm-up delivery fault: " <> show outcome)) + let next = Vector.last events ^. #globalPosition + checked runner (Store.runTransaction (Store.resetSubscriptionCheckpointsTx (Store.SubscriptionName "note-rich-worker" :| []) next)) + drainWarmup runner next + +facts runner = + checked + runner + ( Store.runTransaction + ( Tx.statement + () + ( preparable + "SELECT (SELECT count(*) FROM app.state_note_identity), (SELECT count(*) FROM stream_events WHERE stream_id = 0), (SELECT count(*) FROM app.state_notes), (SELECT stream_version FROM streams WHERE stream_id = 0), COALESCE((SELECT min(last_seen) FROM subscriptions WHERE subscription_name = 'note-rich-worker'),0)" + E.noParams + (D.singleRow ((,,,,) <$> col <*> col <*> col <*> col <*> col)) + ) + ) + ) + where + col = D.column (D.nonNullable D.int8) + +summarize :: Measurements -> Aeson.Value +summarize result = + Aeson.object + [ "attempts" Aeson..= length (samples result), + "accepted" Aeson..= length (filter accepted (samples result)), + "refused" Aeson..= length (filter (not . accepted) (samples result)), + "attemptThroughput" Aeson..= (fromIntegral (length (samples result)) / elapsedSeconds result), + "cohorts" Aeson..= Aeson.object [AesonKey.fromText name Aeson..= latencySummary [latencyMicros sample | sample <- samples result, cohort sample == name] | name <- ["ordinary", "guarded-accepted", "guarded-refused"]] + ] + +latencySummary :: [Double] -> Aeson.Value +latencySummary [] = Aeson.Null +latencySummary values = Aeson.object ["count" Aeson..= length values, "p50" Aeson..= percentile 0.50 values, "p95" Aeson..= percentile 0.95 values, "p99" Aeson..= percentile 0.99 values, "maximum" Aeson..= maximum values] + +percentile :: Double -> [Double] -> Double +percentile fraction values = List.sort values !! min (length values - 1) (floor (fraction * fromIntegral (length values))) diff --git a/bench/Main.hs b/bench/Main.hs --- a/bench/Main.hs +++ b/bench/Main.hs @@ -17,6 +17,7 @@ import Data.UUID qualified as UUID import Effectful (Eff, IOE, (:>)) import Effectful.Error.Static (Error) +import GuardedCatalogBench (runGuardedCatalogEvidence) import InboxDelegatedBench (prepareInboxDelegatedBenchmarks, runInboxDelegatedExplainIfRequested) import Keiki.Core ( Edge (..), @@ -145,7 +146,12 @@ } main :: IO () -main = +main = do + evidence <- runGuardedCatalogEvidence + unless evidence ordinaryMain + +ordinaryMain :: IO () +ordinaryMain = withMigratedSuite \fixture -> withFreshResourceStore fixture \(store, runner) -> do withFreshResourceStore fixture \(readModelStore, readModelRunner) -> do diff --git a/keiro.cabal b/keiro.cabal --- a/keiro.cabal +++ b/keiro.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro -version: 0.19.0.1 +version: 0.20.0.0 synopsis: Event sourcing framework and workflow engine description: A library that composes kiroku, keiki, and shibuya into an @@ -20,6 +20,9 @@ CHANGELOG.md README.md +extra-source-files: + bench/GuardedCatalogBaselineMain.hs + source-repository head type: git location: https://github.com/shinzui/keiro.git @@ -167,13 +170,13 @@ hs-opentelemetry-semantic-conventions >=1.40 && <2, keiki >=0.9 && <0.10, keiki-codec-json >=0.9 && <0.10, - keiro-core ^>=0.19.0.1, - kiroku-store >=0.9.0.1 && <0.10, + keiro-core ^>=0.20.0.0, + kiroku-store ^>=0.11.0.0, lens >=5.2 && <5.4, mmzk-typeid >=0.7 && <0.8, random >=1.2.1 && <1.4, scientific >=0.3 && <0.4, - shibuya-core ^>=0.10.0.0, + shibuya-core ^>=0.10.0.1, stm >=2.5 && <2.6, streamly >=0.11 && <0.12, streamly-core >=0.3 && <0.4, @@ -196,6 +199,10 @@ Compatibility.ReadModel011 ExternalReadSpec GroupRebuildSpec + GuardedCatalogExample + GuardedCatalogFixture + GuardedCatalogSpec + GuardedCatalogVersionedSpec Paths_keiro PreCanonicalRecoverySpec PreimageSpec @@ -233,10 +240,10 @@ keiki, keiki-codec-json, keiro, - keiro-test-support ^>=0.19.0.1, - kiroku-store >=0.9.0.1 && <0.10, + keiro-test-support ^>=0.20.0.0, + kiroku-store ^>=0.11.0.0, process >=1.6 && <1.7, - shibuya-core ^>=0.10.0.0, + shibuya-core ^>=0.10.0.1, stm >=2.5 && <2.6, streamly-core >=0.3 && <0.4, text >=2.1 && <2.2, @@ -269,11 +276,11 @@ keiki >=0.9 && <0.10, keiki-codec-json >=0.9 && <0.10, keiro, - keiro-core ^>=0.19.0.1, - keiro-test-support ^>=0.19.0.1, - kiroku-store >=0.9.0.1 && <0.10, - shibuya-core ^>=0.10.0.0, - shibuya-kiroku-adapter ^>=0.5.1.5, + keiro-core ^>=0.20.0.0, + keiro-test-support ^>=0.20.0.0, + kiroku-store ^>=0.11.0.0, + shibuya-core ^>=0.10.0.1, + shibuya-kiroku-adapter ^>=0.6.0.1, stm >=2.5 && <2.6, streamly-core >=0.3 && <0.4, text >=2.1 && <2.2, @@ -283,9 +290,16 @@ benchmark keiro-bench import: warnings, shared type: exitcode-stdio-1.0 - hs-source-dirs: bench + hs-source-dirs: + bench + test + main-is: Main.hs other-modules: + GuardedCatalogBench + GuardedCatalogExample + GuardedCatalogFixture + GuardedCatalogWorkload InboxDelegatedBench ProducerIdentityBench ReadModelBench @@ -310,12 +324,14 @@ hs-opentelemetry-sdk >=1.0 && <1.1, keiki >=0.9 && <0.10, keiro, - keiro-core ^>=0.19.0.1, - keiro-test-support ^>=0.19.0.1, - kiroku-store >=0.9.0.1 && <0.10, - shibuya-core ^>=0.10.0.0, + keiro-core ^>=0.20.0.0, + keiro-test-support ^>=0.20.0.0, + kiroku-store ^>=0.11.0.0, + process >=1.6 && <1.7, + shibuya-core ^>=0.10.0.1, streamly-core >=0.3 && <0.4, tasty-bench >=0.4, text >=2.1 && <2.2, time >=1.12 && <1.15, uuid >=1.3 && <1.4, + vector >=0.13 && <0.14, diff --git a/retention/Retention/Fixture.hs b/retention/Retention/Fixture.hs --- a/retention/Retention/Fixture.hs +++ b/retention/Retention/Fixture.hs @@ -76,7 +76,7 @@ import Kiroku.Store.Effect (Store) import Kiroku.Store.Effect.Resource (KirokuStoreResource) import Kiroku.Store.Error (StoreError) -import Kiroku.Store.Subscription.Stream (AckItem (..), subscriptionAckStream) +import Kiroku.Store.Subscription.Stream (AckItem (..), mkStreamBufferSize, subscriptionAckStream) import Kiroku.Store.Subscription.Types (SubscriptionConfig) import Kiroku.Store.Subscription.Types qualified as Sub import Kiroku.Store.Types (EventData (..), EventId (..), EventType (..), ExpectedVersion (..), GlobalPosition (..), RecordedEvent (..), StreamName (..), StreamVersion (..)) @@ -394,7 +394,8 @@ ackAdapter :: (IOE :> es) => Store.KirokuStore -> SubscriptionConfig -> Natural -> IO (Adapter es RecordedEvent, IO ()) ackAdapter store config bufferSize = do - (source, cancel) <- subscriptionAckStream store config bufferSize + validatedBufferSize <- either (fail . show) pure (mkStreamBufferSize bufferSize) + (source, cancel) <- subscriptionAckStream store config validatedBufferSize let ingested = fmap (toIngested cancel) (Streamly.morphInner liftIO source) pure ( Adapter {adapterName = "retention-ack-bridge", source = ingested, shutdown = liftIO cancel}, diff --git a/retention/Retention/Legs.hs b/retention/Retention/Legs.hs --- a/retention/Retention/Legs.hs +++ b/retention/Retention/Legs.hs @@ -32,7 +32,7 @@ import Keiro.Test.Postgres (StoreRunner (..)) import Kiroku.Store qualified as Store import Kiroku.Store.Subscription qualified as Subscription -import Kiroku.Store.Subscription.Stream (AckItem (..), subscriptionAckStream) +import Kiroku.Store.Subscription.Stream (AckItem (..), defaultStreamBufferSize, subscriptionAckStream) import Kiroku.Store.Subscription.Types (SubscriptionConfig, SubscriptionName (..), SubscriptionResult (..), SubscriptionTarget (..), defaultSubscriptionConfig) import Kiroku.Store.Transaction qualified as StoreTx import Kiroku.Store.Types (CategoryName (..), EventData (..), EventId (..), EventType (..), ExpectedVersion (..), RecordedEvent, StreamName (..), StreamVersion (..)) @@ -245,7 +245,7 @@ kirokuSubscribeAck config (store, runner) = do appendSignals runner (operationCount config) measured <- measureLegWithSampler config "kiroku-subscribe-ack" \sample -> do - (source, cancel) <- subscriptionAckStream store (sourceConfig "retention-kiroku-ack") 256 + (source, cancel) <- subscriptionAckStream store (sourceConfig "retention-kiroku-ack") defaultStreamBufferSize consumeAck (operationCount config) config.blockSize sample source `finally` cancel postReturn "kiroku-subscribe-ack" (operationCount config) pure measured diff --git a/src/Keiro/ProcessManager.hs b/src/Keiro/ProcessManager.hs --- a/src/Keiro/ProcessManager.hs +++ b/src/Keiro/ProcessManager.hs @@ -56,8 +56,8 @@ -- 'Keiro.Telemetry.kirokuEventBridge' on Kiroku's @eventHandler@ to observe the -- terminal transition. -- --- The adapter's @KirokuAdapterConfig@ does not currently expose @retryPolicy@, so --- that path uses Kiroku's default bound. The sharded path is configurable: +-- The adapter's @KirokuAdapterConfig.retryPolicy@ configures this bound and +-- defaults to Kiroku's five-delivery policy. The sharded path is configurable: -- 'Keiro.Subscription.Shard.Worker.runShardedSubscriptionGroupAck' forwards -- 'Keiro.Subscription.Shard.Worker.ShardedWorkerOptions.retryPolicy' into the -- same Kiroku acknowledgement ladder. A shard handler that dispatches commands @@ -363,6 +363,7 @@ StreamNameTooLong {} -> False DuplicateEvent {} -> False EventAlreadyLinked {} -> False + EventDecodeFailed {} -> False LinkSourceEventMissing {} -> False TransientTransactionFailure {} -> True UnexpectedServerError {} -> False @@ -939,7 +940,7 @@ -- in. On a Kiroku-backed adapter, each 'AckRetry' redelivery is bounded by the -- subscription @RetryPolicy@ (five total deliveries by default). Exhaustion -- dead-letters the source event in @kiroku.dead_letters@ and advances the --- checkpoint; @KirokuAdapterConfig@ does not currently expose that bound. Observe +-- checkpoint; @KirokuAdapterConfig.retryPolicy@ configures that bound. Observe -- the terminal event with 'Keiro.Telemetry.kirokuEventBridge' and replay it with -- @Keiro.DeadLetter.Replay@ when appropriate. runProcessManagerWorker :: diff --git a/src/Keiro/Projection.hs b/src/Keiro/Projection.hs --- a/src/Keiro/Projection.hs +++ b/src/Keiro/Projection.hs @@ -25,6 +25,11 @@ runCommandWithCatalogProjections, DomainProjectionCommandOutcome (..), runDomainCommandWithCatalogProjections, + GuardedCatalogOutcome (..), + CatalogPreconditionContext, + ProjectionWriteBinding (..), + catalogPreconditionBindings, + runDomainCommandWithCatalogProjectionsGuarded, -- * Asynchronous projections AsyncProjection (..), @@ -89,7 +94,8 @@ import Keiro.Projection.Types import Keiro.ReadModel (storeHeadPosition, subscriptionPositionFromInventory) import Keiro.ReadModel.Rebuild.Group - ( ProjectionWriteFence (..), + ( ProjectionWriteBinding (..), + ProjectionWriteFence (..), RebuildRunId, lockProjectionGroupsTx, ) @@ -139,6 +145,21 @@ | DomainProjectionCommandCatalogMismatch !SourceId deriving stock (Generic, Eq, Show) +-- | A precondition refusal proves rollback and carries no accepted command +-- receipt. All ordinary catalog and pure domain outcomes retain their types. +data GuardedCatalogOutcome target co rejection noOp refusal + = CatalogCommandOutcome !(DomainProjectionCommandOutcome target co rejection noOp) + | PreconditionRefused !refusal + deriving stock (Generic, Eq, Show) + +-- | Callback-scoped serving bindings, resolved under all required group locks. +-- Retaining this value does not retain locks or authorize another transaction. +-- Asynchronously maintained targets are not command-time identity authority. +newtype CatalogPreconditionContext = CatalogPreconditionContext [ProjectionWriteBinding] + +catalogPreconditionBindings :: CatalogPreconditionContext -> [ProjectionWriteBinding] +catalogPreconditionBindings (CatalogPreconditionContext bindings) = bindings + -- | Catalog-aware result of one asynchronous projection application. data CatalogAsyncApplyOutcome = CatalogAsyncApplied @@ -290,18 +311,73 @@ toProjectionOutcome = \case DomainSqlCommandSilent outcome -> DomainProjectionCommandApplied outcome DomainSqlCommandCommitted outcome _ -> DomainProjectionCommandApplied outcome - DomainSqlCommandRolledBack fence -> fenceOutcome fence + DomainSqlCommandRolledBack fence -> domainCatalogFenceOutcome fence - fenceOutcome = \case - ProjectionWritesAllowed _ -> - error "runDomainCommandWithCatalogProjections: rolled back with writes allowed" - ProjectionWriteFenced groupId runId -> DomainProjectionCommandFenced groupId runId - ProjectionWriteGroupUnregistered groupId -> DomainProjectionCommandGroupUnregistered groupId - ProjectionServingRevisionUnavailable groupId revisionId -> - DomainProjectionCommandServingRevisionUnavailable groupId revisionId - ProjectionServingBindingInvalid groupId revisionId detail -> - DomainProjectionCommandServingBindingInvalid groupId revisionId detail +domainCatalogFenceOutcome :: ProjectionWriteFence -> DomainProjectionCommandOutcome target co rejection noOp +domainCatalogFenceOutcome = \case + ProjectionWritesAllowed _ -> + error "runDomainCommandWithCatalogProjections: rolled back with writes allowed" + ProjectionWriteFenced groupId runId -> DomainProjectionCommandFenced groupId runId + ProjectionWriteGroupUnregistered groupId -> DomainProjectionCommandGroupUnregistered groupId + ProjectionServingRevisionUnavailable groupId revisionId -> + DomainProjectionCommandServingRevisionUnavailable groupId revisionId + ProjectionServingBindingInvalid groupId revisionId detail -> + DomainProjectionCommandServingBindingInvalid groupId revisionId detail +-- | Check a relational precondition after tentative append and catalog +-- admission, before inline delivery. @Left refusal@ rolls back the whole +-- append transaction; @Right ()@ delivers using the exact protected bindings. +-- Silent domain decisions invoke neither catalog SQL nor the precondition. +-- +-- The callback is trusted, bounded transaction-local application SQL and may +-- run repeatedly on driver retries. Return refusal through 'Either'; never +-- call @Tx.condemn@, control transactions/savepoints/isolation, acquire other +-- stream/group locks, or append recursively. Avoid sequences, session locks, +-- SQL functions with external effects, and other effects surviving rollback. +-- Production identity checks should be bounded indexed reads only. +-- +-- Atomic identity lookup relies on participating writers using the same Kiroku +-- store's global append lock and synchronous authority under ReadCommitted. +-- The callback sees its own tentative append; that is not a committed winner. +-- Every creation path must enforce the invariant. Arbitrary external SQL and +-- lagging async projections do not participate in this guarantee. Keep checks +-- minimal: every post-append statement extends the store-wide append lock. +-- 'RunCommandOptions' bounds OCC retries, not Hasql's transaction retry loop. +runDomainCommandWithCatalogProjectionsGuarded :: + forall phi rs s ci co rejection noOp refusal es. + (HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) => + RunCommandOptions -> + DomainCommandHandler phi rs s ci co rejection noOp -> + Stream (EventStream phi rs s ci co) -> + ci -> + ValidatedProjectionCatalog -> + ProjectionSet co -> + (CatalogPreconditionContext -> Tx.Transaction (Either refusal ())) -> + Eff es (Either CommandError (GuardedCatalogOutcome (EventStream phi rs s ci co) co rejection noOp refusal)) +runDomainCommandWithCatalogProjectionsGuarded options handler targetStream command catalog projectionSet precondition + | Prelude.null groups = + pure (Right (CatalogCommandOutcome (DomainProjectionCommandCatalogMismatch (projectionSet ^. #projectionSource)))) + | otherwise = do + outcome <- runDomainCommandWithSqlEventsControlled options handler targetStream command $ \pairs _ -> do + fence <- lockProjectionGroupsTx catalog groups + case fence of + ProjectionWritesAllowed bindings -> do + decision <- precondition (CatalogPreconditionContext bindings) + case decision of + Left refusal -> pure (RollbackSqlTransaction (Right refusal)) + Right () -> do + deliverCatalogProjectionsTx catalog projectionSet bindings pairs + pure (CommitSqlTransaction (Left fence)) + _ -> pure (RollbackSqlTransaction (Left fence)) + pure (fmap toGuardedOutcome outcome) + where + groups = typedProjectionRebuildGroups catalog projectionSet + toGuardedOutcome = \case + DomainSqlCommandSilent value -> CatalogCommandOutcome (DomainProjectionCommandApplied value) + DomainSqlCommandCommitted value _ -> CatalogCommandOutcome (DomainProjectionCommandApplied value) + DomainSqlCommandRolledBack (Left fence) -> CatalogCommandOutcome (domainCatalogFenceOutcome fence) + DomainSqlCommandRolledBack (Right refusal) -> PreconditionRefused refusal + applyCatalogProjectionsTx :: ValidatedProjectionCatalog -> ProjectionSet co -> @@ -312,9 +388,20 @@ fence <- lockProjectionGroupsTx catalog groups case fence of ProjectionWritesAllowed bindings -> do - traverse_ applyBinding bindings + deliverCatalogProjectionsTx catalog projectionSet bindings pairs pure (CommitSqlTransaction fence) _ -> pure (RollbackSqlTransaction fence) + +-- Group-major, handler-major, event-major order is shared by both runners. +-- All admission checks must finish before calling this delivery-only helper. +deliverCatalogProjectionsTx :: + ValidatedProjectionCatalog -> + ProjectionSet co -> + [ProjectionWriteBinding] -> + [(co, RecordedEvent)] -> + Tx.Transaction () +deliverCatalogProjectionsTx catalog projectionSet bindings pairs = + traverse_ applyBinding bindings where applyBinding binding = case binding ^. #writeRevisionId of diff --git a/src/Keiro/Router.hs b/src/Keiro/Router.hs --- a/src/Keiro/Router.hs +++ b/src/Keiro/Router.hs @@ -631,7 +631,7 @@ -- adapter, 'AckRetry' remains bounded by the subscription @RetryPolicy@ (five -- total deliveries by default). Exhaustion records the source event in -- @kiroku.dead_letters@ with reason kind @max_attempts_exceeded@ and advances the --- checkpoint. @KirokuAdapterConfig@ does not currently expose @retryPolicy@; +-- checkpoint. @KirokuAdapterConfig.retryPolicy@ configures that bound; -- install 'Keiro.Telemetry.kirokuEventBridge' on Kiroku's @eventHandler@ to observe -- the terminal event. The configurable sharded path forwards -- @ShardedWorkerOptions.retryPolicy@ through the same Kiroku ladder; see diff --git a/src/Keiro/Subscription/Shard.hs b/src/Keiro/Subscription/Shard.hs --- a/src/Keiro/Subscription/Shard.hs +++ b/src/Keiro/Subscription/Shard.hs @@ -36,9 +36,13 @@ -- * Lease descriptor ShardLease (..), ShardCountMismatch (..), + ShardResizeActiveLeases (..), + ShardCountResizeReport (..), -- * Ownership operations ensureShards, + resizeShardCount, + resizeShardCountTx, acquireOwnedBuckets, renewOwnedBuckets, relinquish, @@ -54,9 +58,10 @@ import Data.Set (Set) import Data.Set qualified as Set import Data.Time (NominalDiffTime) +import Data.Typeable (cast) import Data.UUID.V4 qualified as UUIDv4 import Effectful (Eff, IOE, (:>)) -import Effectful.Exception (Exception, throwIO) +import Effectful.Exception (Exception (..), throwIO) import Keiro.Prelude import Keiro.Subscription.Shard.Schema ( WorkerId (..), @@ -64,12 +69,17 @@ ensureShardRows, listShardCounts, listShardOwnership, + lockShardNameTx, + lockShardRowsTx, releaseShardsTx, renewLeaseTx, + replaceShardRowsTx, ) import Kiroku.Store.Effect (Store) -import Kiroku.Store.Subscription.Types (SubscriptionName (..)) +import Kiroku.Store.Subscription.Checkpoint (ConsumerGroupResizeReport, resizeConsumerGroupTx) +import Kiroku.Store.Subscription.Types (ConsumerGroupSize, SomeSubscriptionStartupFailure (..), SubscriptionName (..), consumerGroupSizeValue) import Kiroku.Store.Transaction (runTransaction) +import "hasql-transaction" Hasql.Transaction qualified as Tx -- | Mint a fresh per-process 'WorkerId' (a random UUID). freshWorkerId :: (IOE :> es) => Eff es WorkerId @@ -94,8 +104,14 @@ mismatchFound :: ![Int] } deriving stock (Generic, Eq, Show) - deriving anyclass (Exception) +-- | Keiro and Kiroku topology refusals share the released startup-failure parent. +instance Exception ShardCountMismatch where + toException = toException . SomeSubscriptionStartupFailure + fromException exception = do + SomeSubscriptionStartupFailure refusal <- fromException exception + cast refusal + -- | The fair-share claim target: @ceil(N / liveWorkers)@. When @k@ workers are -- live they collectively claim all @N@ buckets and no single worker hogs them. A -- non-positive @liveWorkers@ is treated as one (claim everything). @@ -108,19 +124,72 @@ -- on every worker startup ('ensureShardRows' uses @ON CONFLICT DO NOTHING@). ensureShards :: (Store :> es) => ShardLease -> Eff es () ensureShards lease = do - counts <- runTransaction $ do - ensureShardRows (subscriptionName lease) (shardCount lease) - listShardCounts (subscriptionName lease) - let configured = shardCount lease - found = [n | (n, _) <- counts, n /= configured] - unless (null found) $ - throwIO - ShardCountMismatch - { mismatchSubscriptionName = case subscriptionName lease of - SubscriptionName name -> name, - mismatchConfigured = configured, - mismatchFound = found - } + outcome <- runTransaction $ do + lockShardNameTx (subscriptionName lease) + counts <- listShardCounts (subscriptionName lease) + let configured = shardCount lease + found = [n | (n, _) <- counts, n /= configured] + if null found + then ensureShardRows (subscriptionName lease) configured >> pure (Right ()) + else + pure + ( Left + ShardCountMismatch + { mismatchSubscriptionName = case subscriptionName lease of SubscriptionName name -> name, + mismatchConfigured = configured, + mismatchFound = found + } + ) + either throwIO pure outcome + +-- | Any recorded owner blocks resize, even after expiry. Stop workers and +-- explicitly relinquish ownership before changing hash assignments. +data ShardResizeActiveLeases = ShardResizeActiveLeases + { resizeSubscriptionName :: !SubscriptionName, + resizeOwnedBuckets :: ![Int] + } + deriving stock (Generic, Eq, Show) + deriving anyclass (Exception) + +-- | Both sides of the committed topology change; mixed old shard counts remain +-- visible in the report instead of being collapsed to a guessed single size. +data ShardCountResizeReport = ShardCountResizeReport + { previousShardCounts :: ![(Int, Int)], + resizedShardCount :: !ConsumerGroupSize, + checkpointResize :: !ConsumerGroupResizeReport + } + deriving stock (Generic, Eq, Show) + +-- | Construct the size with 'Kiroku.Store.Subscription.Types.mkConsumerGroupSize' +-- first. Stop all workers, relinquish every lease, then atomically replace lease +-- rows and rewind checkpoints through the released public API in +-- mori://shinzui/kiroku/packages/kiroku-store. Same-size resize equalizes progress +-- too, so a changed hash assignment cannot leave gaps. +resizeShardCount :: (Store :> es) => SubscriptionName -> ConsumerGroupSize -> Eff es ShardCountResizeReport +resizeShardCount name count = do + outcome <- runTransaction (resizeShardCountTx name count) + either throwIO pure outcome + +-- | Compose resize with application SQL. A refused resize performs no writes; +-- any subsequent transaction failure rolls back both checkpoint and lease rows. +resizeShardCountTx :: SubscriptionName -> ConsumerGroupSize -> Tx.Transaction (Either ShardResizeActiveLeases ShardCountResizeReport) +resizeShardCountTx name count = do + rows <- lockShardRowsTx name + let active = [bucket | (bucket, _, owner) <- rows, isJust owner] + if null active + then do + oldCounts <- listShardCounts name + checkpoints <- resizeConsumerGroupTx name count + replaceShardRowsTx name (fromIntegral (consumerGroupSizeValue count)) + pure + ( Right + ShardCountResizeReport + { previousShardCounts = oldCounts, + resizedShardCount = count, + checkpointResize = checkpoints + } + ) + else pure (Left (ShardResizeActiveLeases name active)) -- | One ownership-reconcile pass: in a single transaction, renew the leases this -- worker still holds, then — if it holds fewer than its fair share — claim __one__ diff --git a/src/Keiro/Subscription/Shard/Schema.hs b/src/Keiro/Subscription/Shard/Schema.hs --- a/src/Keiro/Subscription/Shard/Schema.hs +++ b/src/Keiro/Subscription/Shard/Schema.hs @@ -31,6 +31,9 @@ releaseShardsTx, listShardOwnership, listShardCounts, + lockShardNameTx, + lockShardRowsTx, + replaceShardRowsTx, ) where @@ -57,7 +60,47 @@ -- on every worker startup safe; it converges the table to exactly @N@ rows. ensureShardRows :: SubscriptionName -> Int -> Tx.Transaction () ensureShardRows (SubscriptionName name) shardCount = - Tx.statement (name, fromIntegral shardCount) ensureShardRowsStmt + lockShardNameTx (SubscriptionName name) + >> Tx.statement (name, fromIntegral shardCount) ensureShardRowsStmt + +-- | Serialize creation and replacement of the complete lease-row set, including +-- the empty-set case. Claims and renewals remain protected by their row locks. +lockShardNameTx :: SubscriptionName -> Tx.Transaction () +lockShardNameTx (SubscriptionName name) = Tx.statement name lockShardNameStmt + +-- | Lock every existing lease row before inspecting owners or changing topology. +lockShardRowsTx :: SubscriptionName -> Tx.Transaction [(Int, Int, Maybe WorkerId)] +lockShardRowsTx name@(SubscriptionName text) = do + lockShardNameTx name + Tx.statement text lockShardRowsStmt + +-- | Replace stopped lease rows in the caller's transaction. Call only after +-- locking the full set and refusing any nonempty owner. +replaceShardRowsTx :: SubscriptionName -> Int -> Tx.Transaction () +replaceShardRowsTx name@(SubscriptionName text) count = do + Tx.statement text deleteShardRowsStmt + ensureShardRows name count + +lockShardNameStmt :: Statement Text () +lockShardNameStmt = + preparable + "SELECT pg_advisory_xact_lock(hashtext('keiro_subscription_shards'), hashtext($1))" + (E.param (E.nonNullable E.text)) + D.noResult + +lockShardRowsStmt :: Statement Text [(Int, Int, Maybe WorkerId)] +lockShardRowsStmt = + preparable + "SELECT bucket, shard_count, owner_worker_id FROM keiro.keiro_subscription_shards WHERE subscription_name = $1 ORDER BY bucket FOR UPDATE" + (E.param (E.nonNullable E.text)) + (D.rowList ((,,) <$> (fromIntegral <$> D.column (D.nonNullable D.int4)) <*> (fromIntegral <$> D.column (D.nonNullable D.int4)) <*> (fmap WorkerId <$> D.column (D.nullable D.uuid)))) + +deleteShardRowsStmt :: Statement Text () +deleteShardRowsStmt = + preparable + "DELETE FROM keiro.keiro_subscription_shards WHERE subscription_name = $1" + (E.param (E.nonNullable E.text)) + D.noResult -- | Claim up to @targetCount@ buckets that are currently unowned __or__ whose -- lease has expired (@owner_worker_id IS NULL OR lease_expires_at < now@), in one diff --git a/src/Keiro/Subscription/Shard/Worker.hs b/src/Keiro/Subscription/Shard/Worker.hs --- a/src/Keiro/Subscription/Shard/Worker.hs +++ b/src/Keiro/Subscription/Shard/Worker.hs @@ -86,7 +86,7 @@ import Control.Concurrent (forkFinally, killThread, threadDelay) import Control.Concurrent.STM (atomically, putTMVar) -import Control.Exception (SomeAsyncException, SomeException, catch, displayException, finally, fromException, throwIO) +import Control.Exception (Exception, SomeAsyncException, SomeException, catch, displayException, finally, fromException, throwIO) import Control.Monad (forever) import Data.Bifunctor (first) import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef, writeIORef) @@ -111,10 +111,9 @@ ) import Kiroku.Store.Connection (KirokuStore) import Kiroku.Store.Effect (runStoreIO) -import Kiroku.Store.Subscription.Stream (AckItem (..), subscriptionAckStream) +import Kiroku.Store.Subscription.Stream (AckItem (..), mkStreamBufferSize, subscriptionAckStream) import Kiroku.Store.Subscription.Types - ( ConsumerGroup (..), - DeadLetterReason (..), + ( DeadLetterReason (..), RetryDelay (..), RetryPolicy (..), SubscriptionName, @@ -133,6 +132,7 @@ = ShardSnapshotFailed !Text | ShardAcquireFailed !Text | ShardReaderDied !Int !Text + | ShardReaderStartupFailed !Int !Text | ShardEnsureFailed !Text deriving stock (Generic, Eq, Show) @@ -195,6 +195,7 @@ | InvalidShardHandlerRetryDelay !RetryDelay | InvalidShardRetryMaxAttempts !Int deriving stock (Generic, Eq, Show) + deriving anyclass (Exception) -- | Sensible defaults for a sharded worker: 30 s lease, 10 s renew, batch 100, -- buffer 256. Supply the category target and the bucket count @N@. @@ -216,6 +217,7 @@ mkShardedWorkerOptions :: ShardedWorkerOptions -> Either ShardedWorkerConfigError ShardedWorkerOptions mkShardedWorkerOptions opts | opts ^. #shardCount < 1 = Left (InvalidShardCount (opts ^. #shardCount)) + | opts ^. #shardCount > fromIntegral (maxBound :: Int32) = Left (InvalidShardCount (opts ^. #shardCount)) | opts ^. #leaseTtl <= 0 = Left (InvalidShardLeaseTtl (opts ^. #leaseTtl)) | opts ^. #renewInterval <= 0 = Left (InvalidShardRenewInterval (opts ^. #renewInterval)) | opts ^. #leaseTtl <= opts ^. #renewInterval = @@ -311,12 +313,31 @@ Int -> IO RunningReader startReader store lease opts readers handler bucket = do + count <- + either + (const (throwIO (InvalidShardCount (opts ^. #shardCount)))) + pure + (Sub.mkConsumerGroupSize (fromIntegral (opts ^. #shardCount))) + group <- + either + (const (throwIO (InvalidShardCount (opts ^. #shardCount)))) + pure + (Sub.mkConsumerGroup (fromIntegral bucket) count) + batch <- + either + (const (throwIO (InvalidShardBatchSize (opts ^. #batchSize)))) + pure + (Sub.mkBatchSize (opts ^. #batchSize)) + buffer <- + either + (const (throwIO (InvalidShardBufferSize (opts ^. #bufferSize)))) + pure + (mkStreamBufferSize (opts ^. #bufferSize)) stopping <- newIORef False let subConfig = (defaultSubscriptionConfig (lease ^. #subscriptionName) (opts ^. #target) (\_ -> pure Continue)) - { Sub.batchSize = opts ^. #batchSize, - Sub.consumerGroup = - Just (ConsumerGroup {member = fromIntegral bucket, size = fromIntegral (opts ^. #shardCount)}), + { Sub.batchSize = batch, + Sub.consumerGroup = Just group, Sub.retryPolicy = opts ^. #retryPolicy } handleItem item = do @@ -333,7 +354,7 @@ case fromException err of Just async -> throwIO (async :: SomeAsyncException) Nothing -> pure (ShardAckRetry (opts ^. #handlerRetryDelay)) - (stream, cancelAction) <- subscriptionAckStream store subConfig (opts ^. #bufferSize) + (stream, cancelAction) <- subscriptionAckStream store subConfig buffer tid <- forkFinally (Stream.fold (Fold.drainMapM handleItem) stream) $ \result -> do intentional <- readIORef stopping @@ -342,7 +363,11 @@ let reason = case result of Left err -> Text.pack (displayException (err :: SomeException)) Right _ -> "reader stream ended" - reportShardError opts (ShardReaderDied bucket reason) + case result of + Left err | Just refusal <- (fromException err :: Maybe Sub.SomeSubscriptionStartupFailure) -> do + _ <- runStoreIO store (relinquish lease (Set.singleton bucket)) + reportShardError opts (ShardReaderStartupFailed bucket (Text.pack (displayException refusal))) + _ -> reportShardError opts (ShardReaderDied bucket reason) pure ( RunningReader $ do writeIORef stopping True @@ -393,6 +418,7 @@ ShardEventHandler -> IO () runShardedSubscriptionGroupAck store subName opts handler = do + either throwIO (const (pure ())) (mkShardedWorkerOptions opts) worker <- WorkerId <$> UUIDv4.nextRandom let lease = ShardLease diff --git a/test/GuardedCatalogExample.hs b/test/GuardedCatalogExample.hs new file mode 100644 --- /dev/null +++ b/test/GuardedCatalogExample.hs @@ -0,0 +1,50 @@ +-- | Compiled public-API adoption example over the shared note fixture. +module GuardedCatalogExample + ( module GuardedCatalogFixture, + NoteRefusal (..), + EnsureNoteError (..), + ensureStateNoteExample, + noteGuard, + ) +where + +import GuardedCatalogFixture +import Keiro +import Keiro.Prelude +import Keiro.Projection +import Keiro.Test.Postgres (StoreRunner (..)) +import Kiroku.Store qualified as Store +import "hasql-transaction" Hasql.Transaction qualified as Tx + +data NoteRefusal = ExistingNote Text | GuardConfigurationFault Text + deriving stock (Eq, Show) + +data EnsureNoteError + = EnsureStoreFault Store.StoreError + | EnsureCommandFault CommandError + | EnsureGuardFault Text + | EnsureCatalogFault (DomainProjectionCommandOutcome NoteStream NoteEvent Text Text) + deriving stock (Eq, Show) + +-- Only ExistingNote maps to reuse. Framework faults and a missing/unsupported +-- binding remain distinct failures. No rolled-back position escapes this API. +ensureStateNoteExample :: StoreRunner -> Note -> IO (Either EnsureNoteError (Text, Bool)) +ensureStateNoteExample (StoreRunner run) note = do + outcome <- run (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (handler 1) (stream (noteId note)) (CreateNote (notePayload note)) catalog commandSet (noteGuard note)) + pure $ case outcome of + Right (Right (PreconditionRefused (ExistingNote winner))) -> Right (winner, False) + Right (Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}}))) -> Right (noteId note, True) + Left fault -> Left (EnsureStoreFault fault) + Right (Left fault) -> Left (EnsureCommandFault fault) + Right (Right (PreconditionRefused (GuardConfigurationFault detail))) -> Left (EnsureGuardFault detail) + Right (Right (CatalogCommandOutcome other)) -> Left (EnsureCatalogFault other) + +noteGuard :: Note -> CatalogPreconditionContext -> Tx.Transaction (Either NoteRefusal ()) +noteGuard note context = case catalogPreconditionBindings context of + [binding] | binding ^. #writeGroupId == identityGroup && binding ^. #writeRevisionId == Nothing -> + case resolvePhysicalTarget identityTarget (binding ^. #writePhysicalTargets) of + Nothing -> pure (Left (GuardConfigurationFault "identity target missing")) + Just table -> do + winner <- Tx.statement (notePayload note) (identityLookup table) + pure (maybe (Right ()) (Left . ExistingNote) winner) + _ -> pure (Left (GuardConfigurationFault "unsupported identity serving binding")) diff --git a/test/GuardedCatalogFixture.hs b/test/GuardedCatalogFixture.hs new file mode 100644 --- /dev/null +++ b/test/GuardedCatalogFixture.hs @@ -0,0 +1,256 @@ +{-# LANGUAGE DataKinds #-} + +-- | Executable consumer fixture. The identity index is command authority; +-- payload maintenance belongs to the independent asynchronous owner. +module GuardedCatalogFixture where + +import Data.Aeson qualified as Aeson +import Data.ByteString.Lazy qualified as Lazy +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Encoding +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Statement (Statement, preparable) +import Keiki.Core +import Keiki.Core qualified as Keiki +import Keiro +import Keiro.Connection (qualifyTable) +import Keiro.Prelude +import Keiro.Projection +import Keiro.Projection.Catalog (SourceScope (AllStreams)) +import Keiro.ReadModel (QueryCursorAuthority (..), ReadModel, ReadModelBlueprint (..), immediateReadModel) +import Keiro.ReadModel.Rebuild (registerProjectionCatalog) +import Keiro.Test.Postgres (StoreRunner (..)) +import Kiroku.Store qualified as Store +import Kiroku.Store.Subscription.Types (MissingCheckpointPolicy (..)) +import Kiroku.Store.Types (EventType (..), RecordedEvent) +import "hasql-transaction" Hasql.Transaction qualified as Tx +import Prelude + +data NoteCommand = CreateNote Text | ArchiveNote Text | IgnoreNote | RejectNote | UnmatchedNote + deriving stock (Eq, Show) + +data NoteEvent = NoteCreated Text | NoteArchived Text | NoteAudited + deriving stock (Eq, Show) + +data NoteState = Ready deriving stock (Eq, Ord, Show, Enum, Bounded) + +type NoteStream = EventStream (HsPred '[] NoteCommand) '[] NoteState NoteCommand NoteEvent + +-- All identity components are canonical, non-null strings. Archive never +-- removes the mapping. This fixture has no delete or re-anchor command. +data Note = Note + { noteId :: Text, + anchorType :: Text, + anchorId :: Text, + propertyId :: Text, + stateId :: Text, + content :: Text, + tags :: [Text], + provenance :: Text + } + deriving stock (Eq, Show, Generic) + deriving anyclass (Aeson.ToJSON, Aeson.FromJSON) + +notePayload :: Note -> Text +notePayload = Encoding.decodeUtf8 . Lazy.toStrict . Aeson.encode . canonicalNote + +-- The application normalizes the same keys before both lookup and encoding. +canonicalNote :: Note -> Note +canonicalNote note = + note + { anchorType = Text.toCaseFold (Text.strip (anchorType note)), + anchorId = Text.strip (anchorId note), + propertyId = Text.strip (propertyId note), + stateId = Text.strip (stateId note) + } + +exampleNote :: Text -> Text -> Note +exampleNote candidate key = Note candidate "intention" key "status" "ready" (Text.replicate 1024 "x") ["a", "b", "c", "d"] "original-source" + +handler :: Int -> DomainCommandHandler (HsPred '[] NoteCommand) '[] NoteState NoteCommand NoteEvent Text Text +handler batch = DomainCommandHandler (validated batch) $ \context -> + case context ^. #command of + IgnoreNote -> SilentNoOp "unchanged" + RejectNote -> SilentRejected "refused by domain" + _ -> error "eventful command classified as silent" + +validated :: Int -> ValidatedEventStream (HsPred '[] NoteCommand) '[] NoteState NoteCommand NoteEvent +validated batch = either (error . show) id (mkEventStream "state-note" raw) + where + raw = + EventStream + { transducer = + SymTransducer + { edgesOut = + const + [ edge (matchInCtor createCtor) (pack createCtor createdCtor (inpCtor createCtor #payload *: oNil) : replicate (batch - 1) (pack createCtor auditedCtor oNil)), + edge (matchInCtor archiveCtor) [pack archiveCtor archivedCtor (inpCtor archiveCtor #payload *: oNil)], + edge (matchInCtor ignoreCtor) [], + edge (matchInCtor rejectCtor) [] + ], + initial = Ready, + initialRegs = RNil, + isFinal = const False + }, + initialState = Ready, + initialRegisters = RNil, + eventCodec = noteCodec, + resolveStreamName = streamName, + snapshotPolicy = Never, + stateCodec = Nothing + } + edge predicate outputs = Edge predicate UKeep outputs Ready Keiki.Live + +type Payload = '[ '("payload", Text)] + +createCtor, archiveCtor :: InCtor NoteCommand Payload +createCtor = unavailableInCtor "CreateNote" (\case CreateNote p -> Just (RCons Proxy p RNil); _ -> Nothing) (\(RCons _ p RNil) -> CreateNote p) +archiveCtor = unavailableInCtor "ArchiveNote" (\case ArchiveNote p -> Just (RCons Proxy p RNil); _ -> Nothing) (\(RCons _ p RNil) -> ArchiveNote p) + +ignoreCtor, rejectCtor :: InCtor NoteCommand '[] +ignoreCtor = unavailableInCtor "IgnoreNote" (\case IgnoreNote -> Just RNil; _ -> Nothing) (const IgnoreNote) +rejectCtor = unavailableInCtor "RejectNote" (\case RejectNote -> Just RNil; _ -> Nothing) (const RejectNote) + +createdCtor, archivedCtor :: WireCtor NoteEvent (Text, ()) +createdCtor = unavailableWireCtor "NoteCreated" (\case NoteCreated p -> Just (p, ()); _ -> Nothing) (NoteCreated . fst) +archivedCtor = unavailableWireCtor "NoteArchived" (\case NoteArchived p -> Just (p, ()); _ -> Nothing) (NoteArchived . fst) + +auditedCtor :: WireCtor NoteEvent () +auditedCtor = unavailableWireCtor "NoteAudited" (\case NoteAudited -> Just (); _ -> Nothing) (const NoteAudited) + +noteCodec :: Codec NoteEvent +noteCodec = + Codec + { eventTypes = EventType "NoteCreated" :| [EventType "NoteArchived", EventType "NoteAudited"], + eventType = \case NoteCreated _ -> EventType "NoteCreated"; NoteArchived _ -> EventType "NoteArchived"; NoteAudited -> EventType "NoteAudited", + schemaVersion = 1, + encode = \case NoteCreated p -> Aeson.String p; NoteArchived p -> Aeson.String p; NoteAudited -> Aeson.Null, + decode = \(EventType kind) value -> case (kind, value) of + ("NoteCreated", Aeson.String p) -> Right (NoteCreated p) + ("NoteArchived", Aeson.String p) -> Right (NoteArchived p) + ("NoteAudited", _) -> Right NoteAudited + _ -> Left "invalid note event", + upcasters = [] + } + +identityGroup, richGroup :: RebuildGroupId +identityGroup = ident mkRebuildGroupId "note-identity-group" +richGroup = ident mkRebuildGroupId "note-rich-group" + +identityTarget, richTarget :: TargetId +identityTarget = ident mkTargetId "note-identity-target" +richTarget = ident mkTargetId "note-rich-target" + +identityOwner, richOwner :: ProjectionId +identityOwner = ident mkProjectionId "note-identity-owner" +richOwner = ident mkProjectionId "note-rich-owner" + +noteSource :: SourceId +noteSource = ident mkSourceId "note-source" + +noteSubscription :: SubscriptionId +noteSubscription = ident mkSubscriptionId "note-subscription" + +noteDedup :: DedupKeyId +noteDedup = ident mkDedupKeyId "note-dedup" + +ident :: (Show e) => (Text -> Either e a) -> Text -> a +ident constructor = either (error . show) id . constructor + +site :: ClaimSite +site = ident mkClaimSite "guarded-note-example" + +identityTable, richTable :: QualifiedTable +identityTable = QualifiedTable "app" "state_note_identity" +richTable = QualifiedTable "app" "state_notes" + +identityDefinition, richDefinition :: ProjectionDefinition NoteEvent +identityDefinition = ProjectionDefinition identityOwner identityGroup (identityTarget :| []) (replay applyIdentity) (InlineHandler (InlineProjection "note-identity" applyIdentity) site :| []) site +richDefinition = ProjectionDefinition richOwner richGroup (richTarget :| []) (replay applyRich) (AsyncHandler richProjection noteSubscription noteDedup site :| []) site + +commandSet, fullSet :: ProjectionSet NoteEvent +commandSet = ProjectionSet noteSource (identityDefinition :| []) site +fullSet = ProjectionSet noteSource (identityDefinition :| [richDefinition]) site + +richProjection :: AsyncProjection +richProjection = + AsyncProjection + "note-rich" + "note-rich-query" + "note-rich-worker" + ( \recorded -> case decodeNoteRecorded recorded of + ReplayRelevant event -> applyRich event recorded + _ -> pure () + ) + (^. #eventId) + +replay :: (NoteEvent -> RecordedEvent -> Tx.Transaction ()) -> ProjectionReplayPolicy NoteEvent +replay applyEvent = Replayable (replayAdapterFromCodec noteCodec applyEvent) + +decodeNoteRecorded :: RecordedEvent -> ReplayDecodeResult NoteEvent +decodeNoteRecorded = (^. #decodeForReplay) (replayAdapterFromCodec noteCodec applyRich) + +catalogDefinition :: ProjectionCatalog +catalogDefinition = + ProjectionCatalog + { sources = [SourceDeclaration noteSource AllStreams "note-codec-v1" site], + targets = [TargetDeclaration identityTarget identityTable ClearBeforeReplay [] site, TargetDeclaration richTarget richTable ClearBeforeReplay [] site], + rebuildGroups = [RebuildGroupDeclaration identityGroup [identityTarget] [] site, RebuildGroupDeclaration richGroup [richTarget] [] site], + projectionRevisions = [], + externalReadContracts = [], + subscriptions = [SubscriptionDeclaration noteSubscription "note-rich-worker" noteSource FromBeginning site], + dedupKeys = [DedupKeyDeclaration noteDedup "note-rich" site], + queryModels = [SomeQueryModelBinding (QueryModelBinding (ident mkQueryModelId "note-rich-query") richReadModel richGroup [richTarget] site)], + projectionSets = [SomeProjectionSet fullSet] + } + +richReadModel :: ReadModel Text (Maybe Text) +richReadModel = immediateReadModel (ReadModelBlueprint "note-rich-query" "state_notes" "app" 1 "note-rich-v1" (DurableQueryCursor "note-rich-worker") (\candidate -> Tx.statement candidate richLookup)) + +richLookup :: Statement Text (Maybe Text) +richLookup = preparable "SELECT payload::text FROM app.state_notes WHERE note_id = $1" (E.param (E.nonNullable E.text)) (D.rowMaybe (D.column (D.nonNullable D.text))) + +catalog :: ValidatedProjectionCatalog +catalog = case validateProjectionCatalog catalogDefinition of + Success value -> value + Failure diagnostics -> error (show diagnostics) + +setup :: StoreRunner -> IO () +setup (StoreRunner run) = do + run (Store.runTransaction (Tx.sql (Encoding.encodeUtf8 setupSql))) >>= either (fail . show) pure + run (registerProjectionCatalog catalog) >>= either (fail . show) (either (fail . show) (const (pure ()))) + +setupSql :: Text +setupSql = + """ + CREATE SCHEMA app; + CREATE TABLE app.state_note_identity ( + anchor_type text NOT NULL, anchor_id text NOT NULL, + property_id text NOT NULL, state_id text NOT NULL, note_id text NOT NULL, + PRIMARY KEY (anchor_type, anchor_id, property_id, state_id)); + CREATE TABLE app.state_notes (note_id text PRIMARY KEY, payload jsonb NOT NULL, archived boolean NOT NULL DEFAULT false); + CREATE TABLE app.guard_markers (marker text); + """ + +applyIdentity, applyRich :: NoteEvent -> RecordedEvent -> Tx.Transaction () +applyIdentity event _ = case event of + NoteCreated payload -> Tx.statement payload insertIdentity + _ -> pure () +applyRich event _ = case event of + NoteCreated payload -> Tx.statement payload (preparable "INSERT INTO app.state_notes(note_id,payload) SELECT $1::jsonb->>'noteId',$1::jsonb" textParam D.noResult) + NoteArchived candidate -> Tx.statement candidate (preparable "UPDATE app.state_notes SET archived = true WHERE note_id = $1" textParam D.noResult) + NoteAudited -> pure () + +textParam :: E.Params Text +textParam = E.param (E.nonNullable E.text) + +insertIdentity :: Statement Text () +insertIdentity = preparable "INSERT INTO app.state_note_identity SELECT $1::jsonb->>'anchorType',$1::jsonb->>'anchorId',$1::jsonb->>'propertyId',$1::jsonb->>'stateId',$1::jsonb->>'noteId'" textParam D.noResult + +identityLookup :: QualifiedTable -> Statement Text (Maybe Text) +identityLookup table = + preparable + ("SELECT note_id FROM " <> qualifyTable (table ^. #schemaName) (table ^. #tableName) <> " WHERE anchor_type = $1::jsonb->>'anchorType' AND anchor_id = $1::jsonb->>'anchorId' AND property_id = $1::jsonb->>'propertyId' AND state_id = $1::jsonb->>'stateId'") + textParam + (D.rowMaybe (D.column (D.nonNullable D.text))) diff --git a/test/GuardedCatalogSpec.hs b/test/GuardedCatalogSpec.hs new file mode 100644 --- /dev/null +++ b/test/GuardedCatalogSpec.hs @@ -0,0 +1,545 @@ +{-# LANGUAGE BlockArguments #-} + +module GuardedCatalogSpec (spec, checked, withController, script, waitSql, awaitWorker) where + +import Control.Concurrent (forkFinally, killThread, threadDelay) +import Control.Concurrent.MVar +import Control.Exception (bracket, finally) +import Control.Monad (forM_, void) +import Data.Aeson qualified as Aeson +import Data.Either (isLeft) +import Data.IORef (atomicModifyIORef', newIORef) +import Data.Text qualified as Text +import Effectful (Eff, IOE) +import Effectful.Error.Static (Error) +import GuardedCatalogExample qualified as Example +import Hasql.Connection.Settings qualified as Conn +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Pool qualified as Pool +import Hasql.Pool.Config qualified as PoolConfig +import Hasql.Session qualified as Session +import Hasql.Statement (preparable) +import Keiki.Core qualified as Keiki +import Keiro +import Keiro.Prelude +import Keiro.Projection +import Keiro.ReadModel +import Keiro.ReadModel.Rebuild qualified +import Keiro.Test.Postgres +import Kiroku.Store qualified as Store +import Kiroku.Store.Subscription.Types (MissingCheckpointPolicy (..)) +import Kiroku.Store.Types (GlobalPosition (..), StreamName (..), StreamVersion (..)) +import System.Timeout (timeout) +import Test.Hspec hiding (parallel) +import "hasql-transaction" Hasql.Transaction qualified as Tx + +spec :: Fixture -> Spec +spec fixture = describe "guarded catalog commands" $ do + it "uses durable PostgreSQL settings for paired benchmark fixtures" $ + withMigratedBenchmarkSuite \benchmark -> + withFreshResourceStore benchmark \(_, runner) -> + checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT current_setting('fsync'), current_setting('synchronous_commit'), current_setting('full_page_writes'), current_setting('autovacuum')" E.noParams (D.singleRow ((,,,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text)))))) + `shouldReturn` ("on", "on", "on", "on") + for_ [("fresh refusal", 3), ("exhaustion", 0)] \(label, budget) -> + it ("discards an OCC-conflicting decision before " <> label) $ + withFreshResourceStoresSameDatabase fixture \((_, retrying), (_, winner)) -> + occScenario retrying winner (Example.handler 1) budget (if budget == 0 then "exhausted" else "reuse") + it "returns a silent decision after OCC retry without invoking a guard" $ + withFreshResourceStoresSameDatabase fixture \((_, retrying), (_, winner)) -> + occScenario retrying winner onceHandler 3 "noop" + it "opens distinct backends in the same database and schema" $ + withFreshResourceStoresSameDatabase fixture \((_, StoreRunner first), (_, StoreRunner second)) -> do + let facts = + Store.runTransaction $ + Tx.statement () $ + preparable + "SELECT pg_backend_pid(), current_database()::text, current_schema()::text FROM pg_sleep(0.1)" + E.noParams + (D.singleRow ((,,) <$> D.column (D.nonNullable D.int4) <*> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text))) + results <- parallel (first facts) (second facts) + case results of + (Right (pidA, dbA, schemaA), Right (pidB, dbB, schemaB)) -> do + pidA `shouldNotBe` pidB + dbA `shouldBe` dbB + schemaA `shouldBe` schemaB + other -> expectationFailure (show other) + + around (withFreshResourceStore fixture) $ do + it "preserves encode failures before callback or append" \(_, runner) -> do + Example.setup runner + let options = defaultRunCommandOptions & #metadata ?~ Aeson.Null + result <- checked runner (runDomainCommandWithCatalogProjectionsGuarded options (Example.handler 1) (stream "encode-failure") (Example.CreateNote (Example.notePayload (Example.exampleNote "encode-failure" "shared"))) Example.catalog Example.commandSet (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right () :: Either Example.NoteRefusal ()))) + result `shouldBe` Left (EncodeFailed (NonObjectCallerMetadata Aeson.Null)) + counts runner `shouldReturn` (0, 0, 0, 0) + + it "reuses canonical identity keys without replacing the original payload" \(_, runner) -> do + Example.setup runner + let first = (Example.exampleNote "winner" " shared ") {Example.anchorType = " Intention ", Example.propertyId = " status ", Example.stateId = " ready "} + second = Example.exampleNote "duplicate" "shared" + Example.ensureStateNoteExample runner first `shouldReturn` Right ("winner", True) + Example.ensureStateNoteExample runner second `shouldReturn` Right ("winner", False) + counts runner `shouldReturn` (1, 1, 0, 0) + + it "does not advance a snapshot or fixture effects on refusal" \(_, runner) -> do + Example.setup runner + checked runner (Store.runTransaction (Tx.sql "CREATE TABLE app.fixture_outbox(note_id text); CREATE TABLE app.fixture_reactors(note_id text)")) + let raw = unvalidated (Example.validated 1) + codec :: StateCodec (Example.NoteState, Keiki.RegFile '[]) + codec = StateCodec 1 "guarded-ready" "guarded-ready" (const Aeson.Null) (const (Right (Example.Ready, Keiki.RNil))) + validated = either (error . show) id (mkEventStream "guarded-snapshot" raw {snapshotPolicy = Every 1, stateCodec = Just codec}) + handler = (Example.handler 1) {eventStream = validated} + note = Example.exampleNote "snapshot-refusal" "shared" + result <- + checked + runner + ( runDomainCommandWithCatalogProjectionsGuarded + defaultRunCommandOptions + handler + (stream "snapshot-refusal") + (Example.CreateNote (Example.notePayload note)) + Example.catalog + Example.commandSet + ( \_ -> do + Tx.sql "INSERT INTO app.fixture_outbox VALUES ('snapshot-refusal'); INSERT INTO app.fixture_reactors VALUES ('snapshot-refusal')" + pure (Left (Example.ExistingNote "winner")) + ) + ) + result `shouldBe` Right (PreconditionRefused (Example.ExistingNote "winner")) + checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT (SELECT count(*) FROM keiro.keiro_snapshots), (SELECT count(*) FROM app.fixture_outbox), (SELECT count(*) FROM app.fixture_reactors)" E.noParams (D.singleRow ((,,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8)))))) `shouldReturn` (0, 0, 0) + counts runner `shouldReturn` (0, 0, 0, 0) + let acceptedNote = note {Example.noteId = "snapshot-accepted"} + acceptedResult <- checked runner (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions handler (stream "snapshot-accepted") (Example.CreateNote (Example.notePayload acceptedNote)) Example.catalog Example.commandSet (\_ -> pure (Right () :: Either Example.NoteRefusal ()))) + acceptedResult `shouldSatisfy` accepted + checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT count(*) FROM keiro.keiro_snapshots" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))))) `shouldReturn` 1 + + it "rebuilds both catalog owners without invoking the command guard" \(_, runner) -> do + Example.setup runner + checked runner (Store.initializeSubscriptionCheckpoint (Store.SubscriptionName "note-rich-worker") 0 FromBeginning) >>= (`shouldSatisfy` either (const False) (const True)) + let note = Example.exampleNote "winner" "shared" + Example.ensureStateNoteExample runner note `shouldReturn` Right ("winner", True) + for_ [(Example.identityGroup, "identity-rebuild"), (Example.richGroup, "rich-rebuild")] \(group, runName) -> do + let runId = Example.ident Keiro.ReadModel.Rebuild.mkRebuildRunId runName + request = Keiro.ReadModel.Rebuild.RebuildRequest runId "guarded-spec" "rebuild event-derived note owners" (GlobalPosition 0) + options = Keiro.ReadModel.Rebuild.defaultRebuildOptions request + report <- checked runner (Keiro.ReadModel.Rebuild.startCatalogRebuild Example.catalog group options) >>= either (fail . show) pure + report ^. #runStatus `shouldBe` Keiro.ReadModel.Rebuild.RebuildRunPromoted + Example.ensureStateNoteExample runner (note {Example.noteId = "duplicate"}) `shouldReturn` Right ("winner", False) + counts runner `shouldReturn` (1, 1, 1, 0) + payload <- checked runner (Store.runTransaction (Tx.statement "winner" Example.richLookup)) + fmap (Aeson.eitherDecodeStrictText :: Text -> Either String Example.Note) payload `shouldBe` Just (Right note) + + it "preserves early catalog mismatch even for a selected silent command" \(_, runner) -> do + Example.setup runner + let wrong = Example.commandSet & #projectionSource .~ Example.ident mkSourceId "wrong-source" + result <- checked runner (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (Example.handler 1) (stream "silent") Example.IgnoreNote Example.catalog wrong (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right () :: Either Example.NoteRefusal ()))) + result `shouldBe` Right (CatalogCommandOutcome (DomainProjectionCommandCatalogMismatch (Example.ident mkSourceId "wrong-source"))) + + it "preserves sorted group, declared handler, then event delivery order" \(_, runner) -> do + Example.setup runner + let trace label = InlineHandler (InlineProjection label (\event _ -> Tx.statement (label <> ":" <> case event of Example.NoteCreated _ -> "created"; _ -> "audited") (preparable "INSERT INTO app.guard_markers VALUES ($1)" Example.textParam D.noResult))) Example.site + a = Example.identityDefinition & #handlers .~ (trace "a" :| [trace "b"]) + b = Example.richDefinition & #handlers .~ (trace "c" :| []) + set = Example.fullSet & #projectionDefinitions .~ (b :| [a]) + definition = Example.catalogDefinition {projectionSets = [SomeProjectionSet set], subscriptions = [], dedupKeys = [], queryModels = []} + catalog = case validateProjectionCatalog definition of Success value -> value; Failure diagnostics -> error (show diagnostics) + -- Re-registering a changed group slice is deliberately refused. Reset + -- the isolated fixture metadata, then register this trace-only catalog. + checked runner (Store.runTransaction (Tx.sql "DELETE FROM keiro.keiro_read_models; DELETE FROM keiro.keiro_projection_rebuild_groups")) + checked runner (Keiro.ReadModel.Rebuild.registerProjectionCatalog catalog) >>= (`shouldSatisfy` either (const False) (const True)) + result <- + checked + runner + ( runDomainCommandWithCatalogProjectionsGuarded + defaultRunCommandOptions + (Example.handler 4) + (stream "trace") + (Example.CreateNote (Example.notePayload (Example.exampleNote "trace" "trace"))) + catalog + set + ( \context -> do + map (^. #writeGroupId) (catalogPreconditionBindings context) `seq` pure (Right () :: Either Example.NoteRefusal ()) + ) + ) + result `shouldSatisfy` accepted + checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT marker FROM app.guard_markers ORDER BY ctid" E.noParams (D.rowList (D.column (D.nonNullable D.text)))))) + `shouldReturn` concatMap (\label -> [label <> ":created", label <> ":audited", label <> ":audited", label <> ":audited"]) ["a", "b", "c"] + + it "rolls back a refusing guard's marker, events, global links and inline rows" \(_, runner) -> do + Example.setup runner + outcome <- guarded runner "loser" (Example.CreateNote (Example.notePayload (Example.exampleNote "loser" "shared"))) $ \_ -> do + Tx.sql "INSERT INTO app.guard_markers VALUES ('private-refusal')" + pure (Left (Example.ExistingNote "winner")) + outcome `shouldBe` Right (PreconditionRefused (Example.ExistingNote "winner")) + checked runner (Store.readStreamForward (StreamName "loser") (StreamVersion 0) 100) `shouldReturn` mempty + counts runner `shouldReturn` (0, 0, 0, 0) + + it "never invokes a faulting inline handler after refusal" \(_, runner) -> do + Example.setup runner + let projection :: InlineProjection Example.NoteEvent + projection = InlineProjection "note-identity" (\_ _ -> Tx.sql "SELECT 1 / 0") + definition = Example.identityDefinition & #handlers .~ (InlineHandler projection Example.site :| []) + set = Example.commandSet & #projectionDefinitions .~ (definition :| []) + run guard = checked runner (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (Example.handler 1) (stream "handler-fault") (Example.CreateNote (Example.notePayload (Example.exampleNote "handler-fault" "shared"))) Example.catalog set guard) + run (\_ -> pure (Left (Example.ExistingNote "winner"))) `shouldReturn` Right (PreconditionRefused (Example.ExistingNote "winner")) + -- Positive control proves the handler fault is observable if admitted. + run (\_ -> pure (Right ())) >>= (`shouldSatisfy` \case Left StoreFailed {} -> True; _ -> False) + counts runner `shouldReturn` (0, 0, 0, 0) + + it "commits the complete four-event batch only after a successful guard" \(_, runner) -> do + Example.setup runner + let note = Example.exampleNote "winner" "shared" + result <- guarded runner "winner" (Example.CreateNote (Example.notePayload note)) (Example.noteGuard note) + result `shouldSatisfy` accepted + events <- checked runner (Store.readStreamForward (StreamName "winner") (StreamVersion 0) 100) + length events `shouldBe` 4 + counts runner `shouldReturn` (4, 1, 0, 0) + forM_ events \event -> checked runner (Store.runTransaction (applyAsyncProjectionFromCatalog Example.catalog Example.richOwner Example.richProjection event)) >>= (`shouldBe` CatalogAsyncApplied) + counts runner `shouldReturn` (4, 1, 1, 0) + payload <- checked runner (Store.runTransaction (Tx.statement "winner" Example.richLookup)) + fmap (Aeson.eitherDecodeStrictText :: Text -> Either String Example.Note) payload `shouldBe` Just (Right note) + + forM_ [Example.IgnoreNote, Example.RejectNote] \command -> + it ("skips catalog and callback SQL for " <> show command) \(_, runner) -> do + Example.setup runner + checked runner (Store.runTransaction (Tx.sql "DELETE FROM keiro.keiro_read_models; DELETE FROM keiro.keiro_projection_rebuild_groups")) + result <- guarded runner "silent" command (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right ())) + result `shouldSatisfy` \case Right (CatalogCommandOutcome (DomainProjectionCommandApplied _)) -> True; _ -> False + counts runner `shouldReturn` (0, 0, 0, 0) + + it "keeps unmatched commands and genuine guard SQL failures operational" \(_, runner) -> do + Example.setup runner + guarded runner "unmatched" Example.UnmatchedNote (\_ -> pure (Right ())) `shouldReturn` Left CommandRejected + let StoreRunner run = runner + failure <- run (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (Example.handler 1) (stream "fault") (Example.CreateNote (Example.notePayload (Example.exampleNote "fault" "fault"))) Example.catalog Example.commandSet (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right () :: Either Example.NoteRefusal ()))) + failure `shouldSatisfy` \case Right (Left (StoreFailed _)) -> True; _ -> False + counts runner `shouldReturn` (0, 0, 0, 0) + + forM_ ["missing", "rebuilding", "failed"] \status -> + it ("gives a " <> status <> " catalog precedence over duplicate refusal") \(_, runner) -> do + Example.setup runner + checked + runner + ( Store.runTransaction + ( Tx.sql + ( case status of + "missing" -> "DELETE FROM keiro.keiro_projection_rebuild_groups WHERE group_id = 'note-identity-group'" + "rebuilding" -> "UPDATE keiro.keiro_projection_rebuild_groups SET status = 'rebuilding', active_run_id = 'test-run', writes_allowed = false, reads_allowed = false WHERE group_id = 'note-identity-group'" + _ -> "UPDATE keiro.keiro_projection_rebuild_groups SET status = 'failed', active_run_id = 'test-run', writes_allowed = false, reads_allowed = false WHERE group_id = 'note-identity-group'" + ) + ) + ) + result <- guarded runner "fenced" (Example.CreateNote (Example.notePayload (Example.exampleNote "fenced" "shared"))) (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Left (Example.ExistingNote "winner"))) + let expected = if status == "missing" then DomainProjectionCommandGroupUnregistered Example.identityGroup else DomainProjectionCommandFenced Example.identityGroup (either (error . show) id (Keiro.ReadModel.Rebuild.mkRebuildRunId "test-run")) + result `shouldBe` Right (CatalogCommandOutcome expected) + counts runner `shouldReturn` (0, 0, 0, 0) + + it "uses an inline-only set while the rich group is fenced, and rejects a broad set" \(_, runner) -> do + Example.setup runner + checked runner (Store.runTransaction (Tx.sql "UPDATE keiro.keiro_projection_rebuild_groups SET status = 'failed', active_run_id = 'test-run', reads_allowed = false, writes_allowed = false WHERE group_id = 'note-rich-group'")) + let note = Example.exampleNote "winner" "shared" + faulting :: InlineProjection Example.NoteEvent + faulting = InlineProjection "note-identity" (\_ _ -> Tx.sql "SELECT 1 / 0") + firstDefinition = Example.identityDefinition & #handlers .~ (InlineHandler faulting Example.site :| []) + broadSet = Example.fullSet & #projectionDefinitions .~ (firstDefinition :| [Example.richDefinition]) + Example.ensureStateNoteExample runner note `shouldReturn` Right ("winner", True) + Example.ensureStateNoteExample runner (note {Example.noteId = "loser"}) `shouldReturn` Right ("winner", False) + result <- checked runner (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (Example.handler 1) (stream "broad") (Example.CreateNote (Example.notePayload (Example.exampleNote "broad" "other"))) Example.catalog broadSet (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right () :: Either Example.NoteRefusal ()))) + result `shouldBe` Right (CatalogCommandOutcome (DomainProjectionCommandFenced Example.richGroup (either (error . show) id (Keiro.ReadModel.Rebuild.mkRebuildRunId "test-run")))) + counts runner `shouldReturn` (1, 1, 0, 0) + + it "retains the original payload through duplicate calls, archive and replay" \(_, runner) -> do + Example.setup runner + let winner = Example.exampleNote "winner" "shared" + loser = winner {Example.noteId = "loser", Example.content = "replacement", Example.tags = ["replacement"], Example.provenance = "replacement"} + Example.ensureStateNoteExample runner winner `shouldReturn` Right ("winner", True) + archived <- guardedWithBatch 1 runner "winner" (Example.ArchiveNote "winner") (\_ -> pure (Right ())) + archived `shouldSatisfy` accepted + Example.ensureStateNoteExample runner loser `shouldReturn` Right ("winner", False) + events <- checked runner (Store.readStreamForward (StreamName "winner") (StreamVersion 0) 100) + forM_ events \event -> checked runner (Store.runTransaction (applyAsyncProjectionFromCatalog Example.catalog Example.richOwner Example.richProjection event)) >>= (`shouldBe` CatalogAsyncApplied) + payload <- checked runner (Store.runTransaction (Tx.statement "winner" Example.richLookup)) + fmap (Aeson.eitherDecodeStrictText :: Text -> Either String Example.Note) payload `shouldBe` Just (Right winner) + checked runner (Store.runTransaction (Tx.sql "TRUNCATE app.state_note_identity, app.state_notes")) + forM_ events \event -> case Example.decodeNoteRecorded event of + ReplayRelevant decoded -> checked runner (Store.runTransaction (Example.applyIdentity decoded event >> Example.applyRich decoded event)) + other -> expectationFailure (show other) + Example.ensureStateNoteExample runner loser `shouldReturn` Right ("winner", False) + counts runner `shouldReturn` (2, 1, 1, 0) + + forM_ [False, True] \reverseStart -> + it ("races independent stores with a paused rich worker, reverse=" <> show reverseStart) $ + withFreshResourceStoresSameDatabase fixture \((_, first), (_, second)) -> do + Example.setup first + let a = Example.exampleNote "a" "shared" + b = (Example.exampleNote "b" "shared") {Example.content = "different", Example.tags = ["different"], Example.provenance = "different"} + start <- newEmptyMVar + let call runner note = readMVar start >> Example.ensureStateNoteExample runner note + results <- + if reverseStart + then parallel (putMVar start () >> call second b) (call first a) + else parallel (putMVar start () >> call first a) (call second b) + case results of + (Right (winner, changed), Right (same, otherChanged)) -> do + winner `shouldBe` same + changed `shouldBe` not otherChanged + counts first `shouldReturn` (1, 1, 0, 0) + let original = if winner == "a" then a else b + events <- checked first (Store.readStreamForward (StreamName winner) (StreamVersion 0) 10) + forM_ events \event -> checked first (Store.runTransaction (applyAsyncProjectionFromCatalog Example.catalog Example.richOwner Example.richProjection event)) >>= (`shouldBe` CatalogAsyncApplied) + counts first `shouldReturn` (1, 1, 1, 0) + payload <- checked first (Store.runTransaction (Tx.statement winner Example.richLookup)) + fmap (Aeson.eitherDecodeStrictText :: Text -> Either String Example.Note) payload `shouldBe` Just (Right original) + other -> expectationFailure (show other) + + forM_ [False, True] \rollbackFirst -> + it ("observes an append lock wait then " <> if rollbackFirst then "rolls back the first writer" else "commits the first writer") $ + withFreshResourceStoresSameDatabaseWithConnection fixture \connection ((_, first), (_, second)) -> + withController connection \controller -> do + Example.setup first + script controller "SELECT pg_advisory_lock(306)" + firstDone <- newEmptyMVar + secondDone <- newEmptyMVar + let a = Example.exampleNote "a" "shared" + b = Example.exampleNote "b" "shared" + firstGuard context = do + Tx.sql "SET LOCAL application_name = 'keiro-306-held-guard'; SELECT pg_advisory_xact_lock(306)" + if rollbackFirst then pure (Left (Example.GuardConfigurationFault "test rollback")) else Example.noteGuard a context + firstThread <- forkFinally (guarded first "a" (Example.CreateNote (Example.notePayload a)) firstGuard) (putMVar firstDone) + let cleanup = script controller "SELECT pg_advisory_unlock_all()" >> killThread firstThread + flip finally cleanup do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE application_name = 'keiro-306-held-guard' AND wait_event_type = 'Lock')" + secondThread <- forkFinally (Example.ensureStateNoteExample second b) (putMVar secondDone) + flip finally (killThread secondThread) do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_stat_activity blocked, pg_stat_activity holder WHERE holder.application_name = 'keiro-306-held-guard' AND holder.pid = ANY(pg_blocking_pids(blocked.pid)))" + script controller "SELECT pg_advisory_unlock(306)" + firstResult <- awaitWorker firstDone + secondResult <- awaitWorker secondDone + if rollbackFirst + then do + firstResult `shouldBe` Right (PreconditionRefused (Example.GuardConfigurationFault "test rollback")) + secondResult `shouldBe` Right ("b", True) + counts first `shouldReturn` (1, 1, 0, 0) + else do + firstResult `shouldSatisfy` accepted + secondResult `shouldBe` Right ("a", False) + counts first `shouldReturn` (4, 1, 0, 0) + + around (withFreshResourceStore fixture) $ do + forM_ ["40001", "40P01"] \sqlState -> + it ("re-evaluates the callback after a one-time driver " <> sqlState <> " retry") \(_, runner) -> do + Example.setup runner + checked runner (Store.runTransaction (Tx.sql "CREATE SEQUENCE app.driver_attempts")) + let note = Example.exampleNote "retry" "shared" + retryGuard context = do + decision <- Example.noteGuard note context + attempt <- Tx.statement () (preparable "SELECT nextval('app.driver_attempts')" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))) + if attempt == 1 + then do + Tx.sql (if sqlState == "40001" then "DO $$ BEGIN RAISE EXCEPTION 'test retry' USING ERRCODE = '40001'; END $$" else "DO $$ BEGIN RAISE EXCEPTION 'test retry' USING ERRCODE = '40P01'; END $$") + pure decision + else pure (Left (Example.ExistingNote "fresh-result")) + result <- guarded runner "retry" (Example.CreateNote (Example.notePayload note)) retryGuard + result `shouldBe` Right (PreconditionRefused (Example.ExistingNote "fresh-result")) + checked runner (Store.runTransaction (Tx.statement () (preparable "SELECT last_value FROM app.driver_attempts" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))))) `shouldReturn` 2 + counts runner `shouldReturn` (0, 0, 0, 0) + + it "waits after commit and retries the same receipt when the rich view catches up" \(_, runner) -> do + Example.setup runner + checked runner (Store.initializeSubscriptionCheckpoint (Store.SubscriptionName "note-rich-worker") 0 FromBeginning) >>= (`shouldSatisfy` either (const False) (const True)) + let note = Example.exampleNote "winner" "shared" + result <- guarded runner "winner" (Example.CreateNote (Example.notePayload note)) (Example.noteGuard note) + receipt <- case result of + Right (CatalogCommandOutcome (DomainProjectionCommandApplied outcome)) -> pure (outcome ^. #result) + other -> expectationFailure (show other) >> error "unreachable" + let options = PositionWaitOptions (receipt ^. #globalPosition) 20000 1000 + checked runner (runQueryWithFreshness Nothing (WaitForPosition options) Example.richReadModel "winner") >>= (`shouldSatisfy` \case Left ReadModelWaitTimeout {} -> True; _ -> False) + events <- checked runner (Store.readStreamForward (StreamName "winner") (StreamVersion 0) 100) + forM_ events \event -> checked runner (Store.runTransaction (applyAsyncProjectionFromCatalog Example.catalog Example.richOwner Example.richProjection event)) >>= (`shouldBe` CatalogAsyncApplied) + checked runner (Store.runTransaction (Store.resetSubscriptionCheckpointsTx (Store.SubscriptionName "note-rich-worker" :| []) (fromMaybe (error "accepted receipt lacks position") (receipt ^. #globalPosition)))) + ready <- checked runner (runQueryWithFreshness Nothing (WaitForPosition options) Example.richReadModel "winner") + ready `shouldSatisfy` \case Right (Just _) -> True; _ -> False + counts runner `shouldReturn` (4, 1, 1, 0) + checked runner (Store.runTransaction (Tx.sql "UPDATE keiro.keiro_read_models SET status='rebuilding' WHERE name='note-rich-query'")) + checked runner (runQueryWithFreshness Nothing (WaitForPosition options) Example.richReadModel "winner") `shouldReturn` Left (ReadModelNotLive "note-rich-query" Rebuilding) + -- A committed receipt remains valid even when its view is unavailable. + Example.ensureStateNoteExample runner (note {Example.noteId = "later-duplicate"}) `shouldReturn` Right ("winner", False) + + it "reads a newly committed winner after a driver retry, under the append protocol" $ + withFreshResourceStoresSameDatabaseWithConnection fixture \connection ((_, retrying), (_, other)) -> + withController connection \controller -> do + Example.setup retrying + -- Test-only backend instrumentation survives rollback. A BEFORE + -- STATEMENT trigger gates the retry before it can acquire append row + -- locks, permitting a third transaction to commit a real winner. + checked + retrying + ( Store.runTransaction + ( Tx.sql + """ + CREATE SEQUENCE app.fresh_driver_attempts; + CREATE FUNCTION app.gate_driver_retry() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'advisory' + AND pid = pg_backend_pid() AND objid = 30601 AND granted) THEN + PERFORM pg_advisory_xact_lock(30602); + END IF; + RETURN NULL; + END $$; + CREATE TRIGGER gate_driver_retry BEFORE UPDATE ON streams + FOR EACH STATEMENT EXECUTE FUNCTION app.gate_driver_retry(); + """ + ) + ) + script controller "SELECT pg_advisory_lock(30602)" + done <- newEmptyMVar + let attempted = Example.exampleNote "retry-candidate" "shared" + winner = Example.exampleNote "winner" "shared" + guard context = do + result <- Example.noteGuard attempted context + count <- Tx.statement () (preparable "SELECT nextval('app.fresh_driver_attempts')" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))) + if count == 1 + then do + Tx.sql "SELECT pg_advisory_lock(30601); DO $$ BEGIN RAISE EXCEPTION 'one-time retry' USING ERRCODE = '40001'; END $$" + pure result + else do + Tx.sql "SELECT pg_advisory_unlock(30601)" + pure result + worker <- forkFinally (guardedWithBatch 1 retrying "retry-candidate" (Example.CreateNote (Example.notePayload attempted)) guard) (putMVar done) + flip finally (script controller "SELECT pg_advisory_unlock_all()" >> killThread worker) do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND objid = 30602 AND NOT granted)" + Example.ensureStateNoteExample other winner `shouldReturn` Right ("winner", True) + script controller "SELECT pg_advisory_unlock(30602)" + awaitWorker done `shouldReturn` Right (PreconditionRefused (Example.ExistingNote "winner")) + counts retrying `shouldReturn` (1, 1, 0, 0) + +withController connection = + bracket + (Pool.acquire (PoolConfig.settings [PoolConfig.staticConnectionSettings (Conn.connectionString connection), PoolConfig.size 1])) + Pool.release + +script controller sql = Pool.use controller (Session.script sql) >>= either (fail . show) pure + +waitSql controller sql = do + result <- timeout (5 * 1000000) loop + case result of Just () -> pure (); Nothing -> fail ("SQL observation timed out: " <> Text.unpack sql) + where + loop = do + ready <- Pool.use controller (Session.statement () (preparable sql E.noParams (D.singleRow (D.column (D.nonNullable D.bool))))) >>= either (fail . show) pure + unless ready (threadDelay 10000 >> loop) + +awaitWorker done = + timeout (20 * 1000000) (takeMVar done) >>= \case + Just (Right value) -> pure value + Just (Left err) -> fail (show err) + Nothing -> fail "worker completion timed out" + +type Outcome = Either CommandError (GuardedCatalogOutcome Example.NoteStream Example.NoteEvent Text Text Example.NoteRefusal) + +guarded :: StoreRunner -> Text -> Example.NoteCommand -> (CatalogPreconditionContext -> Tx.Transaction (Either Example.NoteRefusal ())) -> IO Outcome +guarded = guardedWithBatch 4 + +guardedWithBatch :: Int -> StoreRunner -> Text -> Example.NoteCommand -> (CatalogPreconditionContext -> Tx.Transaction (Either Example.NoteRefusal ())) -> IO Outcome +guardedWithBatch batch runner candidate command guard = checked runner (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (Example.handler batch) (stream candidate) command Example.catalog Example.commandSet guard) + +checked :: StoreRunner -> Eff '[Store.Store, Error Store.StoreError, Store.KirokuStoreResource, IOE] a -> IO a +checked (StoreRunner run) action = run action >>= either (fail . show) pure + +accepted :: Outcome -> Bool +accepted = \case Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}})) -> True; _ -> False + +counts :: StoreRunner -> IO (Int64, Int64, Int64, Int64) +counts runner = + checked + runner + ( Store.runTransaction + ( Tx.statement + () + ( preparable + "SELECT (SELECT count(*) FROM stream_events WHERE stream_id = 0), (SELECT count(*) FROM app.state_note_identity), (SELECT count(*) FROM app.state_notes), (SELECT count(*) FROM app.guard_markers)" + E.noParams + (D.singleRow ((,,,) <$> col <*> col <*> col <*> col)) + ) + ) + ) + where + col = D.column (D.nonNullable D.int8) + +occScenario retrying winner handler budget expected = do + Example.setup retrying + checked retrying (Store.runTransaction (Tx.sql "CREATE SEQUENCE app.occ_guard_reads")) + hydrated <- newEmptyMVar + release <- newEmptyMVar + attempts <- newIORef (0 :: Int) + done <- newEmptyMVar + let note = Example.exampleNote "same-stream" "shared" + hook = do + attempt <- atomicModifyIORef' attempts (\n -> (n + 1, n)) + when (attempt == 0) (putMVar hydrated () >> takeMVar release) + options = defaultRunCommandOptions & #retryLimit .~ budget & #beforeAppend .~ hook + guard context = do + _ <- Tx.statement () (preparable "SELECT nextval('app.occ_guard_reads')" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))) + Example.noteGuard note context + run options runner callback = checked runner (runDomainCommandWithCatalogProjectionsGuarded options handler (stream "same-stream") (Example.CreateNote (Example.notePayload note)) Example.catalog Example.commandSet callback) + label = \case + Left RetryExhausted {} -> "exhausted" + Right (PreconditionRefused (Example.ExistingNote "same-stream")) -> "reuse" + Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainNoOp "already-created"})) -> "noop" + Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}})) -> "accepted" + other -> Text.pack (show other) + worker <- forkFinally (label <$> run options retrying guard) (putMVar done) + flip finally (void (tryPutMVar release ()) >> killThread worker) do + timeout (5 * 1000000) (takeMVar hydrated) >>= (`shouldBe` Just ()) + label <$> run defaultRunCommandOptions winner (Example.noteGuard note) `shouldReturn` "accepted" + putMVar release () + awaitWorker done `shouldReturn` expected + checked retrying (Store.runTransaction (Tx.statement () (preparable "SELECT CASE WHEN is_called THEN last_value ELSE 0 END FROM app.occ_guard_reads" E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))))) + `shouldReturn` (if expected == "reuse" then 1 else 0) + counts retrying `shouldReturn` (1, 1, 0, 0) + +data OnceState = Uncreated | Stored deriving stock (Eq, Ord, Show, Enum, Bounded) + +onceHandler :: DomainCommandHandler (Keiki.HsPred '[] Example.NoteCommand) '[] OnceState Example.NoteCommand Example.NoteEvent Text Text +onceHandler = DomainCommandHandler validated (const (SilentNoOp "already-created")) + where + original = unvalidated (Example.validated 1) + validated = + either + (error . show) + id + ( mkEventStream + "note-once" + EventStream + { transducer = + Keiki.SymTransducer + { edgesOut = \case + Uncreated -> [Keiki.Edge predicate update outputs Stored mode | Keiki.Edge predicate update outputs _ mode <- take 1 (Keiki.edgesOut (original ^. #transducer) Example.Ready)] + Stored -> [Keiki.Edge (Keiki.matchInCtor Example.createCtor) Keiki.UKeep [] Stored Keiki.Live], + initial = Uncreated, + initialRegs = Keiki.RNil, + isFinal = const False + }, + initialState = Uncreated, + initialRegisters = Keiki.RNil, + eventCodec = Example.noteCodec, + resolveStreamName = streamName, + snapshotPolicy = Never, + stateCodec = Nothing + } + ) + +parallel :: IO a -> IO b -> IO (a, b) +parallel first second = do + a <- newEmptyMVar + b <- newEmptyMVar + threadA <- forkFinally first (putMVar a) + threadB <- forkFinally second (putMVar b) + result <- + timeout (20 * 1000000) ((,) <$> takeMVar a <*> takeMVar b) + `finally` (killThread threadA >> killThread threadB) + case result of + Just (Right x, Right y) -> pure (x, y) + Just other -> fail ("worker fault: " <> showErrors other) + Nothing -> fail "same-database workers timed out" + where + showErrors (a, b) = either show (const "success") a <> "/" <> either show (const "success") b diff --git a/test/GuardedCatalogVersionedSpec.hs b/test/GuardedCatalogVersionedSpec.hs new file mode 100644 --- /dev/null +++ b/test/GuardedCatalogVersionedSpec.hs @@ -0,0 +1,236 @@ +{-# LANGUAGE BlockArguments #-} + +module GuardedCatalogVersionedSpec (spec) where + +import Control.Concurrent (forkFinally, killThread) +import Control.Concurrent.MVar +import Control.Exception (finally) +import Data.Map.Strict qualified as Map +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Encoding +import Data.Time (secondsToDiffTime) +import GuardedCatalogExample qualified as Example +import GuardedCatalogSpec (awaitWorker, checked, script, waitSql, withController) +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Statement (preparable) +import Keiro +import Keiro.Connection (qualifyTable) +import Keiro.Prelude +import Keiro.Projection +import Keiro.ReadModel.Rebuild +import Keiro.Test.Postgres +import Kiroku.Store qualified as Store +import Kiroku.Store.HistoryRetention +import Test.Hspec +import "hasql-transaction" Hasql.Transaction qualified as Tx + +spec :: Fixture -> Spec +spec fixture = describe "guarded catalog commands serving revisions" $ do + for_ ["v1", "v2"] \candidateVersion -> do + it ("protects a writer before promotion to " <> Text.unpack candidateVersion) $ + withFreshResourceStoresSameDatabaseWithConnection fixture \connection ((_, writer), (_, promoter)) -> + withController connection \controller -> do + (catalog, request) <- setupVersioned writer candidateVersion + -- Catch up without fencing the currently writable serving revision. + _ <- checked writer (beginVersionedRebuild catalog request) >>= right + _ <- driveToWritableReady writer catalog (request ^. #rebuildRunId) + script controller "SELECT pg_advisory_lock(306)" + done <- newEmptyMVar + let note = Example.exampleNote "protected" "shared" + guard context = do + Tx.sql "SET LOCAL application_name = 'keiro-306-versioned-writer'; SELECT pg_advisory_xact_lock(306)" + versionedGuard "v1" note context + worker <- forkFinally (runGuarded writer catalog note guard) (putMVar done) + flip finally (script controller "SELECT pg_advisory_unlock_all()" >> killThread worker) do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE application_name = 'keiro-306-versioned-writer' AND wait_event_type = 'Lock')" + promotedDone <- newEmptyMVar + promotion <- forkFinally (driveToPromoted promoter catalog (request ^. #rebuildRunId)) (putMVar promotedDone) + flip finally (killThread promotion) do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_stat_activity blocked, pg_stat_activity holder WHERE holder.application_name = 'keiro-306-versioned-writer' AND holder.pid = ANY(pg_blocking_pids(blocked.pid)))" + script controller "SELECT pg_advisory_unlock(306)" + awaitWorker done >>= (`shouldSatisfy` isAccepted) + _ <- awaitWorker promotedDone + runGuarded writer catalog (note {Example.noteId = "later"}) (versionedGuard candidateVersion note) + `shouldReturn` Right (PreconditionRefused (Example.ExistingNote "protected")) + + it ("admits a writer coherently after waiting for promotion to " <> Text.unpack candidateVersion) $ + withFreshResourceStoresSameDatabaseWithConnection fixture \connection ((_, writer), (_, promoter)) -> + withController connection \controller -> do + (catalog, request) <- setupVersioned writer candidateVersion + _ <- checked writer (beginVersionedRebuild catalog request) >>= right + _ <- driveToReady writer catalog (request ^. #rebuildRunId) + prepared <- checked writer (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= right + prepared ^. #promotionPrepared `shouldBe` True + -- Hold a real target lock on the third connection. Promotion takes + -- its group lock first, then waits for the target; the writer's + -- admission statement must cross the committed metadata transition. + script controller "BEGIN; LOCK TABLE app.state_note_identity IN ACCESS SHARE MODE" + promotionDone <- newEmptyMVar + promotion <- forkFinally (checked promoter (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= right) (putMVar promotionDone) + flip finally (script controller "ROLLBACK" >> killThread promotion) do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE relation = 'app.state_note_identity'::regclass AND mode = 'AccessExclusiveLock' AND NOT granted)" + writerDone <- newEmptyMVar + let note = Example.exampleNote "late" "shared" + worker <- forkFinally (runGuarded writer catalog note (versionedGuard candidateVersion note)) (putMVar writerDone) + flip finally (killThread worker) do + waitSql controller "SELECT EXISTS(SELECT 1 FROM pg_stat_activity blocked, pg_locks holder WHERE holder.relation = 'app.state_note_identity'::regclass AND holder.mode = 'AccessExclusiveLock' AND NOT holder.granted AND holder.pid = ANY(pg_blocking_pids(blocked.pid)))" + script controller "COMMIT" + report <- awaitWorker promotionDone + report ^. #phase `shouldBe` VersionedPromoted + result <- awaitWorker writerDone + result `shouldSatisfy` \case + Right (CatalogCommandOutcome (DomainProjectionCommandApplied _)) -> True + Right (CatalogCommandOutcome DomainProjectionCommandServingBindingInvalid {}) -> True + Right (CatalogCommandOutcome DomainProjectionCommandServingRevisionUnavailable {}) -> True + _ -> False + -- A fresh statement after promotion must resolve the new binding. + final <- runGuarded writer catalog note (versionedGuard candidateVersion note) + if isAccepted result + then final `shouldBe` Right (PreconditionRefused (Example.ExistingNote "late")) + else final `shouldSatisfy` isAccepted + -- A falsely valid old-generation binding would let the first + -- call insert into a retired table, then let the fresh call + -- create again in the serving table. Do not accept that pair. + events <- checked writer (Store.readStreamForward (Store.StreamName "late") (Store.StreamVersion 0) 100) + length events `shouldBe` 1 + + around (withFreshResourceStore fixture) $ do + it "refuses an unknown serving revision before invoking the guard" \(_, runner) -> do + (catalog, request) <- setupVersioned runner "v2" + _ <- checked runner (beginVersionedRebuild catalog request) >>= right + let oldOnly = fromDefinition (versionedDefinition [revision "v2"]) + note = Example.exampleNote "unknown" "shared" + result <- runGuarded runner oldOnly note (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right ())) + result `shouldBe` Right (CatalogCommandOutcome (DomainProjectionCommandServingRevisionUnavailable Example.identityGroup (revisionIdFor "v1"))) + + it "refuses an incomplete serving generation map before invoking the guard" \(_, runner) -> do + (catalog, request) <- setupVersioned runner "v2" + _ <- checked runner (beginVersionedRebuild catalog request) >>= right + checked runner (Store.runTransaction (Tx.sql "UPDATE keiro.keiro_projection_target_generations SET lifecycle = 'retired', retired_at = now() WHERE lifecycle = 'serving'")) + let note = Example.exampleNote "invalid" "shared" + result <- runGuarded runner catalog note (\_ -> Tx.sql "SELECT 1 / 0" >> pure (Right ())) + result `shouldSatisfy` \case Right (CatalogCommandOutcome DomainProjectionCommandServingBindingInvalid {}) -> True; _ -> False + +type Outcome = Either CommandError (GuardedCatalogOutcome Example.NoteStream Example.NoteEvent Text Text Example.NoteRefusal) + +runGuarded :: StoreRunner -> ValidatedProjectionCatalog -> Example.Note -> (CatalogPreconditionContext -> Tx.Transaction (Either Example.NoteRefusal ())) -> IO Outcome +runGuarded runner catalog note guard = checked runner (runDomainCommandWithCatalogProjectionsGuarded defaultRunCommandOptions (Example.handler 1) (stream (Example.noteId note)) (Example.CreateNote (Example.notePayload note)) catalog Example.commandSet guard) + +isAccepted :: Outcome -> Bool +isAccepted = \case Right (CatalogCommandOutcome (DomainProjectionCommandApplied DomainCommandOutcome {decision = DomainAccepted {}})) -> True; _ -> False + +versionedGuard :: Text -> Example.Note -> CatalogPreconditionContext -> Tx.Transaction (Either Example.NoteRefusal ()) +versionedGuard expected note context = case catalogPreconditionBindings context of + [binding] | binding ^. #writeRevisionId == Just (revisionIdFor expected) -> case resolvePhysicalTarget Example.identityTarget (binding ^. #writePhysicalTargets) of + Just table -> do + -- SQL for v2 actually requires its additional column. A stale serving + -- revision or retired/candidate table cannot silently pass this check. + when (expected == "v2") (Tx.sql (Encoding.encodeUtf8 ("SELECT v2_marker FROM " <> qualified table <> " LIMIT 1"))) + winner <- Tx.statement (Example.notePayload note) (Example.identityLookup table) + pure (maybe (Right ()) (Left . Example.ExistingNote) winner) + Nothing -> pure (Left (Example.GuardConfigurationFault "target absent")) + _ -> pure (Left (Example.GuardConfigurationFault "wrong serving revision")) + +setupVersioned :: StoreRunner -> Text -> IO (ValidatedProjectionCatalog, VersionedRebuildRequest) +setupVersioned runner candidate = do + checked runner (Store.runTransaction (Tx.sql (Encoding.encodeUtf8 Example.setupSql))) + let catalog = fromDefinition (versionedDefinition [revision "v1", revision "v2"]) + _ <- checked runner (registerProjectionCatalog catalog) >>= right + let targets = either (error . show) id (mkPhysicalTargets [Example.identityTarget] (Map.singleton Example.identityTarget Example.identityTable)) + runId = Example.ident mkRebuildRunId "guarded-versioned-run" + request = + VersionedRebuildRequest + runId + Example.identityGroup + (revisionIdFor "v1") + (revisionIdFor candidate) + targets + ApplicationProvisioned + 100 + 100 + 5000 + 1000000 + (HistoryRetentionLeaseRequest (Example.ident mkHistoryRetentionLeaseOwner "guarded-versioned-test") (Example.ident mkHistoryRetentionLeaseReason "guarded serving generation proof") (either (error . show) id (mkHistoryRetentionLeaseDuration (secondsToDiffTime 600)))) + "guarded-spec" + "serving guard proof" + pure (catalog, request) + +versionedDefinition :: [ProjectionRevision] -> ProjectionCatalog +versionedDefinition revisions = Example.catalogDefinition {projectionRevisions = revisions} + +fromDefinition :: ProjectionCatalog -> ValidatedProjectionCatalog +fromDefinition definition = case validateProjectionCatalog definition of Success value -> value; Failure diagnostics -> error (show diagnostics) + +revisionIdFor :: Text -> ProjectionRevisionId +revisionIdFor version = Example.ident mkProjectionRevisionId ("note-" <> version) + +revision :: Text -> ProjectionRevision +revision version = + ProjectionRevision + (revisionIdFor version) + Example.identityGroup + (Map.singleton Example.identityTarget provisioner) + [RevisionLiveHandler ("note-live-" <> version) 1 (RevisionInlineDelivery Example.identityOwner "note-identity") [Example.identityTarget] applyRecorded] + [RevisionReplayAdapter ("note-replay-" <> version) 1 [Example.identityTarget] (\targets recorded -> applyRecorded targets recorded >> pure (Right True))] + [] + [] + Example.site + where + provisioner = + TargetProvisioner + ("note-provision-" <> version) + 1 + (TargetSchemaVersion version) + ("note-shape-" <> version) + ( \context -> do + Tx.sql (Encoding.encodeUtf8 ("CREATE TABLE " <> qualified (context ^. #stagingTable) <> " (LIKE " <> qualified (context ^. #servingTable) <> " INCLUDING ALL)")) + when (version == "v2") (Tx.sql (Encoding.encodeUtf8 ("ALTER TABLE " <> qualified (context ^. #stagingTable) <> " ADD COLUMN v2_marker text"))) + ) + ("note-validator-" <> version) + 1 + (Just validate) + [] + validate context = do + let table = context ^. #stagingTable + oid <- Tx.statement () (preparable ("SELECT '" <> table ^. #schemaName <> "." <> table ^. #tableName <> "'::regclass::oid::bigint") E.noParams (D.singleRow (D.column (D.nonNullable D.int8)))) + pure (Right (TargetSchemaEvidence oid ("note-shape-" <> version) [] ("note-columns-" <> version))) + applyRecorded targets recorded = case Example.decodeNoteRecorded recorded of + ReplayRelevant (Example.NoteCreated payload) -> case resolvePhysicalTarget Example.identityTarget targets of + Just table -> + Tx.statement + payload + ( preparable + ("INSERT INTO " <> qualified table <> "(anchor_type, anchor_id, property_id, state_id, note_id) SELECT $1::jsonb->>'anchorType',$1::jsonb->>'anchorId',$1::jsonb->>'propertyId',$1::jsonb->>'stateId',$1::jsonb->>'noteId'") + Example.textParam + D.noResult + ) + Nothing -> error "validated target disappeared" + _ -> pure () + +qualified :: QualifiedTable -> Text +qualified table = qualifyTable (table ^. #schemaName) (table ^. #tableName) + +right :: (Show e) => Either e a -> IO a +right = either (fail . show) pure + +driveToReady runner catalog runId = go (20 :: Int) + where + go 0 = fail "cutover did not become ready" + go n = do + report <- checked runner (resumeVersionedRebuild catalog runId) >>= right + if report ^. #phase == VersionedCutoverReplaying && all (\source -> source ^. #exhaustedThrough == Just (source ^. #targetPosition)) (report ^. #sources) then pure report else go (n - 1) + +driveToPromoted runner catalog runId = go (20 :: Int) + where + go 0 = fail "promotion did not finish" + go n = do + report <- checked runner (resumeVersionedRebuild catalog runId) >>= right + if report ^. #phase == VersionedPromoted then pure report else go (n - 1) + +driveToWritableReady runner catalog runId = go (20 :: Int) + where + go 0 = fail "ordinary replay did not finish" + go n = do + report <- checked runner (resumeVersionedRebuild catalog runId) >>= right + if report ^. #phase == VersionedReplayRunning && all (\source -> source ^. #exhaustedThrough == Just (source ^. #targetPosition)) (report ^. #sources) then pure report else go (n - 1) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -44,6 +44,8 @@ import ExternalReadSpec qualified import GHC.Conc (ThreadStatus (..), listThreads, threadStatus) import GroupRebuildSpec qualified +import GuardedCatalogSpec qualified +import GuardedCatalogVersionedSpec qualified import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable) @@ -204,14 +206,19 @@ import Keiro.Stream qualified as Stream import Keiro.Subscription.Shard ( ShardCountMismatch (..), + ShardCountResizeReport (..), ShardLease (..), + ShardResizeActiveLeases (..), WorkerId (..), ensureShards, fairShareTarget, + resizeShardCount, + resizeShardCountTx, ) import Keiro.Subscription.Shard.Schema ( claimShardsTx, ensureShardRows, + listShardCounts, listShardOwnership, releaseShardsTx, renewLeaseTx, @@ -350,6 +357,7 @@ import Kiroku.Store.Effect (Store) import Kiroku.Store.SQL qualified as KirokuSQL import Kiroku.Store.Subscription.Stream (AckItem (..), subscriptionAckStream) +import Kiroku.Store.Subscription.Stream qualified as KirokuStream import Kiroku.Store.Subscription.Types ( SubscriptionName (..), SubscriptionTarget (..), @@ -437,6 +445,8 @@ CatalogEvolutionSpec.spec fixture CatalogOperationsSpec.spec fixture GroupRebuildSpec.spec fixture + GuardedCatalogSpec.spec fixture + GuardedCatalogVersionedSpec.spec fixture ExternalReadSpec.spec fixture VersionedTargetPostgresSpec.spec fixture VersionedRebuildSpec.spec fixture @@ -727,7 +737,7 @@ parseEither (withObject "OrderPlaced" (.: "n")) (ackEvent item ^. #payload) - (stream0, cancelStream) <- subscriptionAckStream store subConfig 4 + (stream0, cancelStream) <- subscriptionAckStream store subConfig (either (error . show) id (KirokuStream.mkStreamBufferSize 4)) ( do (first, stream1) <- pull "initial poison delivery" stream0 ackAttempt first `shouldBe` 0 @@ -5673,6 +5683,10 @@ (RetryDelay 5) (StoreFailed (Store.TransientTransactionFailure "40P01" "deadlock detected")) `shouldBe` AckRetry (RetryDelay 5) + ackForCommandError (RetryDelay 5) (StoreFailed (Store.EventDecodeFailed (Store.DecodeFailure (EventId sampleUuid) "bad payload"))) + `shouldSatisfy` \case + AckHalt (HaltFatal _) -> True + _ -> False ackForCommandError (RetryDelay 5) (StoreFailed (Store.UnexpectedServerError "XX000" "boom")) `shouldSatisfy` \case AckHalt (HaltFatal _) -> True @@ -11845,6 +11859,8 @@ shardOpts = defaultShardedWorkerOptions (Category (CategoryName "orders")) 4 it "validates sharded worker options before startup" $ \_store -> do shouldBeRight_ (mkShardedWorkerOptions shardOpts) + mkShardedWorkerOptions (shardOpts & #shardCount .~ (fromIntegral (maxBound :: Int32) + 1)) + `shouldBeLeft` InvalidShardCount (fromIntegral (maxBound :: Int32) + 1) mkShardedWorkerOptions (shardOpts & #shardCount .~ 0) `shouldBeLeft` InvalidShardCount 0 mkShardedWorkerOptions (shardOpts & #leaseTtl .~ 0) @@ -11937,7 +11953,109 @@ `shouldThrow` \case ShardCountMismatch name configured found -> name == "orders-shard" && configured == 6 && found == [4] + Right counts <- Store.runStoreIO store (Store.runTransaction (listShardCounts subName)) + counts `shouldBe` [(4, 4)] + Store.runStoreIO store (ensureShards lease6) + `shouldThrow` (\(_ :: KirokuSub.SomeSubscriptionStartupFailure) -> True) + describe "Shard count resize" $ around (withFreshStore fixture) $ do + let name = SubscriptionName "resize-orders" + size n = either (error . show) id (KirokuSub.mkConsumerGroupSize n) + positions store = do + Right inventory <- Store.runStoreIO store Store.subscriptionCheckpointInventory + pure [row ^. #checkpointPosition | row <- Vector.toList (inventory ^. #checkpoints), row ^. #subscriptionName == name] + ownership store = do + Right rows <- Store.runStoreIO store (Store.runTransaction (listShardOwnership name)) + pure rows + seedSkewed store = do + Right () <- Store.runStoreIO store (Store.runTransaction (Tx.sql createShardSinkSql >> ensureShardRows name 3)) + _ <- seedOrders store 24 5 + for_ [0 .. 2] $ \member -> do + calls <- newIORef (0 :: Int) + let group = either (error . show) id (KirokuSub.mkConsumerGroup member (size 3)) + config = + ( KirokuSub.defaultSubscriptionConfig + name + (Category (CategoryName "orders")) + ( \event -> do + count <- atomicModifyIORef' calls (\n -> (n + 1, n + 1)) + sinkHandler store member event + pure $ if count > fromIntegral member + 1 then KirokuSub.Stop else KirokuSub.Continue + ) + ) + { KirokuSub.consumerGroup = Just group, + KirokuSub.batchSize = either (error . show) id (KirokuSub.mkBatchSize 1) + } + handle <- Store.subscribe store config + finished <- timeout 10000000 (KirokuSub.wait handle) `finally` (KirokuSub.cancel handle) + case finished of + Just (Right ()) -> pure () + other -> expectationFailure ("seed worker failed: " <> show other) + before <- positions store + length before `shouldBe` 3 + length (Set.fromList before) `shouldSatisfy` (> 1) + pure before + resize store count = do + Right report <- Store.runStoreIO store (resizeShardCount name (size count)) + pure report + + it "refuses recorded owners, including expired leases, without changing either table" $ \store -> do + before <- seedSkewed store + let oldTime = UTCTime (ModifiedJulianDay 60000) 0 + Right claimed <- Store.runStoreIO store (Store.runTransaction (claimShardsTx name (WorkerId sampleUuid) 1 oldTime 1)) + leases <- ownership store + Store.runStoreIO store (resizeShardCount name (size 2)) + `shouldThrow` (\(ShardResizeActiveLeases actual buckets) -> actual == name && buckets == claimed) + positions store `shouldReturn` before + ownership store `shouldReturn` leases + + it "shrinks and grows stopped groups at the old minimum without obsolete members" $ \store -> do + before <- seedSkewed store + report <- resize store 2 + report ^. #previousShardCounts `shouldBe` [(3, 3)] + report ^. #checkpointResize . #resumePosition `shouldBe` minimum before + positions store `shouldReturn` replicate 2 (minimum before) + map (\(bucket, _, _) -> bucket) <$> ownership store `shouldReturn` [0, 1] + _ <- resize store 5 + positions store `shouldReturn` replicate 5 (minimum before) + map (\(bucket, _, _) -> bucket) <$> ownership store `shouldReturn` [0 .. 4] + + it "equalizes same-size assignments and is idempotent on repeat" $ \store -> do + before <- seedSkewed store + _ <- resize store 3 + first <- positions store + first `shouldBe` replicate 3 (minimum before) + _ <- resize store 3 + positions store `shouldReturn` first + + it "rolls back checkpoint and lease changes when surrounding SQL fails" $ \store -> do + before <- seedSkewed store + leases <- ownership store + outcome <- Store.runStoreIO store $ Store.runTransaction $ do + resized <- resizeShardCountTx name (size 2) + case resized of + Left refusal -> error (show refusal) + Right _ -> Tx.sql "SELECT 1 / 0" + outcome `shouldSatisfy` \case Left _ -> True; Right _ -> False + positions store `shouldReturn` before + ownership store `shouldReturn` leases + + it "delivers every seeded event after a changed hash assignment" $ \store -> do + _ <- seedSkewed store + _ <- resize store 2 + let options = + defaultShardedWorkerOptions (Category (CategoryName "orders")) 2 + & #renewInterval + .~ 0.1 + & #leaseTtl + .~ 3 + & #batchSize + .~ 1 + worker <- forkIO (runShardedSubscriptionGroup store name options (sinkHandler store 9)) + complete <- waitUntilSinkCount store 120 15000000 `finally` killThread worker + complete `shouldBe` True + shardSinkCount store `shouldReturn` 120 + describe "Sharded subscription single worker" $ around (withFreshStore fixture) $ do -- EP-51 M3: one process owning all N buckets drains a seeded category exactly -- once. The sink is idempotent on event_id, so "count == total" proves every @@ -17681,7 +17799,7 @@ Just (Just itemAndRest) -> pure itemAndRest Just Nothing -> fail (label <> ": subscription ended early") Nothing -> fail (label <> ": timed out waiting for delivery") - (stream0, cancelStream) <- subscriptionAckStream storeHandle subConfig 4 + (stream0, cancelStream) <- subscriptionAckStream storeHandle subConfig (either (error . show) id (KirokuStream.mkStreamBufferSize 4)) ( do (first, stream1) <- pull "source delivery" stream0 atomically $ @@ -18112,8 +18230,8 @@ upsertSubscriptionCursorStmt = preparable """ - INSERT INTO subscriptions (subscription_name, stream_name, last_seen) - VALUES ($1, '$all', $2) + INSERT INTO subscriptions (subscription_name, target_kind, last_seen) + VALUES ($1, 'all', $2) ON CONFLICT (subscription_name, consumer_group_member) DO UPDATE SET last_seen = EXCLUDED.last_seen, updated_at = now() @@ -18128,8 +18246,8 @@ upsertSubscriptionCursorMemberStmt = preparable """ - INSERT INTO subscriptions (subscription_name, stream_name, consumer_group_member, consumer_group_size, last_seen) - VALUES ($1, '$all', $2, 2, $3) + INSERT INTO subscriptions (subscription_name, target_kind, consumer_group_member, consumer_group_size, last_seen) + VALUES ($1, 'all', $2, 2, $3) ON CONFLICT (subscription_name, consumer_group_member) DO UPDATE SET last_seen = EXCLUDED.last_seen, updated_at = now() diff --git a/test/VersionedRebuildSpec.hs b/test/VersionedRebuildSpec.hs --- a/test/VersionedRebuildSpec.hs +++ b/test/VersionedRebuildSpec.hs @@ -2880,8 +2880,8 @@ upsertSubscriptionCursorStmt = preparable """ - INSERT INTO subscriptions (subscription_name, stream_name, last_seen) - VALUES ($1, '$all', $2) + INSERT INTO subscriptions (subscription_name, target_kind, last_seen) + VALUES ($1, 'all', $2) ON CONFLICT (subscription_name, consumer_group_member) DO UPDATE SET last_seen = EXCLUDED.last_seen, updated_at = now()