FailureInjection.hs

{- | Failure-injection tests for kiroku-store.

Each test deliberately disrupts a runtime dependency (the LISTEN
connection, the connection pool) and asserts the system recovers per its
documented contract.

Currently covers:

  * F14: Listener killed during a window where an event is appended.
    The event is still delivered after the listener reconnects, because
    the publisher re-queries the database from its last position when
    the next NOTIFY arrives — no safety-poll wait needed.

F15 (pool exhaustion → 'PoolAcquisitionTimeout') is deferred until the
'acquisitionTimeout' field is exposed in 'ConnectionSettings'. The
mapping itself ('Kiroku.Store.Error.mapUsageError'
@AcquisitionTimeoutUsageError -> PoolAcquisitionTimeout@) is
straightforward; without a way to set a sub-second acquisition timeout
from the test, the test would either hang (default unbounded timeout)
or require modifying the production type. See the EP-6 plan's M2 notes.
-}
module Test.FailureInjection (spec) where

import Control.Concurrent.STM (atomically, newTVarIO, readTVar, retry, writeTVar)
import Control.Lens ((&), (.~), (^.))
import Data.Aeson qualified as Aeson
import Data.Generics.Labels ()
import Data.IORef (modifyIORef', newIORef, readIORef)
import Kiroku.Store
import Test.Helpers (
    makeEvent,
    terminateBackend,
    waitForListenerPid,
    waitForListenerPidNotEqual,
    waitWithTimeout,
    withTestStoreSettings,
 )
import Test.Hspec

spec :: Spec
spec = describe "kiroku-store failure injection" $ do
    -- F14 — Listener killed during a window where an event is
    -- appended. The publisher's broadcast loop pulls events directly
    -- from PostgreSQL when the next NOTIFY arrives, so the
    -- down-window event is delivered as soon as a post-reconnect
    -- event triggers a tick. No 30-second safety-poll wait required.
    it "delivers events appended during listener-down window after reconnect (F14)" $ \() -> do
        ref <- newIORef ([] :: [RecordedEvent])
        countVar <- newTVarIO (0 :: Int)
        let handler' evt = do
                modifyIORef' ref (evt :)
                n <- atomically $ do
                    c <- readTVar countVar
                    let c' = c + 1
                    writeTVar countVar c'
                    pure c'
                if n >= 3 then pure Stop else pure Continue
        let cfg =
                SubscriptionConfig
                    { name = SubscriptionName "f14-down-window"
                    , target = AllStreams
                    , handler = handler'
                    , batchSize = 100
                    , queueCapacity = 16
                    , overflowPolicy = DropSubscription
                    , consumerGroup = Nothing
                    , consumerGroupGuard = False
                    , retryPolicy = defaultRetryPolicy
                    , eventTypeFilter = AllEventTypes
                    , selector = Nothing
                    }
        withTestStoreSettings (& #eventHandler .~ Nothing) $ \store -> do
            handle <- subscribe store cfg
            -- Pre-event: confirm the subscription is alive and the
            -- listener has reached pg_stat_activity.
            Right _ <- runStoreIO store $ appendToStream (StreamName "f14-pre") NoStream [makeEvent "Pre" (Aeson.object [])]
            atomically $ do
                c <- readTVar countVar
                if c >= 1 then pure () else retry
            -- Find listener pid, terminate it. The listener loop
            -- will reconnect with a fresh pid.
            --
            -- NOTE: pg_terminate_backend is the "administrator command"
            -- that makes PostgreSQL report
            --   FATAL: terminating connection due to administrator command
            -- on the targeted backend. The kiroku-listener's libpq socket
            -- (inside hasql-notifications' waitForNotifications) receives
            -- that as it dies, so that FATAL line on stderr during this
            -- test is EXPECTED — it is the fault we are injecting, not a
            -- connection leak or a teardown-ordering bug. (log_min_messages
            -- is already PANIC, so this is the client seeing a server-
            -- initiated termination, not server-side logging.)
            pid1 <- waitForListenerPid (store ^. #pool) 5_000_000
            terminateBackend (store ^. #pool) pid1
            -- Append the during-down-window event. No NOTIFY reaches
            -- the publisher right now because the listener is
            -- disconnected.
            Right _ <- runStoreIO store $ appendToStream (StreamName "f14-during") NoStream [makeEvent "During" (Aeson.object [])]
            -- Wait for the listener to come back with a fresh pid.
            _ <- waitForListenerPidNotEqual (store ^. #pool) pid1 15_000_000
            -- Append the post-reconnect event. The new NOTIFY
            -- triggers the publisher to fetch from its last position
            -- — which picks up both the during-down event AND this
            -- one.
            Right _ <- runStoreIO store $ appendToStream (StreamName "f14-after") NoStream [makeEvent "After" (Aeson.object [])]
            result <- waitWithTimeout 30_000_000 handle
            case result of
                Left timeout -> expectationFailure timeout
                Right (Left err) -> expectationFailure ("Subscription failed: " <> show err)
                Right (Right ()) -> pure ()
            collected <- readIORef ref
            length collected `shouldBe` 3