keiro-pgmq-0.6.0.0: test/Main.hs
{-# LANGUAGE BlockArguments #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedRecordDot #-}
-- | End-to-end integration test for @keiro-pgmq@.
--
-- Uses @keiro-test-support@ to start one suite-level PostgreSQL server, installs
-- the PGMQ schema into the migrated template database, then gives every example a
-- fresh cloned database. The tests drive the package's public API against that
-- isolated database: 'enqueue' puts work on a queue, 'runJobOnce' drains it, and
-- we read the queue back through @pgmq-effectful@'s 'queueMetrics' to prove that a
-- @Done@ handler deletes the message, a @Retry@ handler leaves it, and a @Dead@
-- handler (or an undecodable payload) routes it to the dead-letter queue.
module Main (main) where
import Control.Concurrent (threadDelay)
import Control.Exception (bracket, throwIO)
import Data.Aeson (FromJSON, ToJSON, Value (..), object, parseJSON, toJSON, (.=))
import Data.Aeson.Key qualified as Key
import Data.Aeson.KeyMap qualified as KeyMap
import Data.Aeson.Types (parseEither)
import Data.Either (isRight)
import Data.Foldable (toList, traverse_)
import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)
import Data.Int (Int32, Int64)
import Data.List (find)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Text (Text)
import Data.Text qualified as Text
import Effectful (Eff, IOE, liftIO, (:>))
import Effectful.Error.Static (Error)
import Effectful.Reader.Static (Reader)
import GHC.Generics (Generic)
import Hasql.Connection.Settings qualified as Conn
import Hasql.Decoders qualified as Decoders
import Hasql.Encoders qualified as Encoders
import Hasql.Pool (Pool)
import Hasql.Pool qualified as Pool
import Hasql.Pool.Config qualified as Pool.Config
import Hasql.Session qualified as Session
import Hasql.Statement qualified as Statement
import Keiro.Codec (EventType (..))
import Keiro.Codec qualified as CoreCodec
import Keiro.PGMQ
import Keiro.Test.Postgres qualified as Postgres
import OpenTelemetry.Attributes (Attribute (..), Attributes, PrimitiveAttribute (..), lookupAttribute)
import OpenTelemetry.Context qualified as Ctxt
import OpenTelemetry.Context.ThreadLocal qualified as CtxtLocal
import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter)
import OpenTelemetry.Processor.Span (SpanProcessor)
import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C
import OpenTelemetry.Trace.Core (Event (..), ImmutableSpan (..), SpanHot (..))
import OpenTelemetry.Trace.Core qualified as OTel
import OpenTelemetry.Trace.Id.Generator.Default (defaultIdGenerator)
import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)
import Pgmq.Config.Types qualified as Config
import Pgmq.Effectful (Message (..), MessageBody (..), Pgmq, QueueMetrics (..), ReadMessage (..), SendMessage (..))
import Pgmq.Effectful qualified as Pgmq
import Pgmq.Migration qualified as Migration
import Pgmq.Types (QueueName, parseQueueName, queueNameToText)
import Shibuya.Adapter.Pgmq (PgmqAdapterEnv)
import Shibuya.App (AppHandle, ShutdownConfig (..), SupervisionStrategy (IgnoreFailures), stopAppGracefully)
import Shibuya.Telemetry.Effect (Tracing)
import System.Timeout (timeout)
import Test.Hspec
-- | A sample job payload defined entirely in the test.
data Ping = Ping
{ message :: Text,
count :: Int
}
deriving stock (Eq, Show, Generic)
deriving anyclass (ToJSON, FromJSON)
-- | The effect stack 'runJobEff' interprets.
type Stack = '[Reader PgmqAdapterEnv, Pgmq, Tracing, Error PgmqRuntimeError, IOE]
main :: IO ()
main = do
-- PGMQ's schema is installed by appending pgmq-migration's native component to
-- the suite's framework plan, so one pg-migrate ledger owns kiroku, keiro, and
-- pgmq together.
pgmq <- either (fail . show) pure Migration.pgmqMigrations
Postgres.withMigratedSuiteWith [pgmq] \fixture ->
hspec $
describe "Keiro.PGMQ" $
around (Postgres.withFreshDatabase fixture) spec
withPool :: Text -> (Pool -> IO a) -> IO a
withPool connStr =
bracket
( Pool.acquire $
Pool.Config.settings
[Pool.Config.staticConnectionSettings (Conn.connectionString connStr)]
)
Pool.release
-- | Count the rows currently in a DLQ's archive table @pgmq.a_<dlqPhysical>@ via a
-- raw @hasql@ session. PGMQ exposes no "read the archive" function, so retention is
-- proven with plain SQL. The queue name is sanitized to @[a-z0-9_]@ by 'queueRef',
-- so interpolating it into the table identifier is safe here.
archiveCount :: Text -> Text -> IO Int64
archiveCount connStr dlqPhysical =
withPool connStr $ \pool -> do
let sql = "SELECT count(*) FROM pgmq.a_" <> dlqPhysical
session =
Session.statement () $
Statement.preparable
sql
Encoders.noParams
(Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))
result <- Pool.use pool session
either (\e -> fail ("archive count failed: " <> show e)) pure result
-- | Run a @Stack@ action against a fresh 'JobRuntime' (no tracer), failing the
-- test on any PGMQ runtime error.
runDb :: Text -> Eff Stack a -> IO a
runDb connStr act =
withJobRuntime connStr Nothing $ \rt -> do
res <- runJobEff rt act
either (\e -> fail ("PGMQ runtime error: " <> show e)) pure res
-- | A job over 'Ping' with a distinct queue name per test (avoids collisions).
mkJob :: Text -> Job Ping
mkJob name =
Job
{ jobName = name,
jobQueue = queueRef name,
jobCodec = aesonJobCodec,
jobPolicy = defaultRetryPolicy
}
versionedPingCodec :: CoreCodec.Codec Ping
versionedPingCodec =
CoreCodec.Codec
{ eventTypes = EventType "ping" :| [],
eventType = \_ -> EventType "ping",
schemaVersion = 2,
encode = toJSON,
decode = \_ value ->
case parseEither parseJSON value of
Left err -> Left (Text.pack err)
Right ping -> Right ping,
upcasters =
[ ( 1,
\_ value ->
case value of
String msg ->
Right $
object
[ "message" .= msg,
"count" .= (1 :: Int)
]
_ -> Left "expected v1 string payload"
)
]
}
-- | Total number of messages currently on a queue (visible or not).
queueLen :: QueueName -> Eff Stack Int64
queueLen q = do
metrics <- Pgmq.queueMetrics q
pure metrics.queueLength
-- | Look up a queue by physical name in a 'Pgmq.listQueues' result and report
-- whether it is unlogged. 'Nothing' means the queue was not found.
queueIsUnlogged :: QueueName -> [Pgmq.Queue] -> Maybe Bool
queueIsUnlogged qn queues =
fmap (.isUnlogged) (find (\q -> q.name == qn) queues)
readOneIsEmpty :: QueueName -> Eff Stack Bool
readOneIsEmpty q = do
messages <-
Pgmq.readMessage
ReadMessage
{ queueName = q,
delay = 30,
batchSize = Just 1,
conditional = Nothing
}
pure (null messages)
-- | Read up to @n@ messages back off a queue (making them invisible for 30 s).
readMessages :: QueueName -> Int32 -> Eff Stack [Message]
readMessages q n = do
messages <-
Pgmq.readMessage
ReadMessage
{ queueName = q,
delay = 30,
batchSize = Just n,
conditional = Nothing
}
pure (toList messages)
-- | Look up a single key in a @Maybe Value@ header object.
headerKey :: Text -> Maybe Value -> Maybe Value
headerKey k = \case
Just (Object o) -> KeyMap.lookup (Key.fromText k) o
_ -> Nothing
-- | A real tracer provider with the W3C Trace Context propagator and a
-- non-dummy id generator, so an active span produces a @traceparent@ on injection.
-- Dummy ids cannot encode a valid @traceparent@, which is why the default id
-- generator is wired in explicitly.
mkW3CProvider :: [SpanProcessor] -> IO OTel.TracerProvider
mkW3CProvider processors =
OTel.createTracerProvider
processors
OTel.emptyTracerProviderOptions
{ OTel.tracerProviderOptionsIdGenerator = defaultIdGenerator,
OTel.tracerProviderOptionsPropagators = W3C.w3cTraceContextPropagator
}
-- | A W3C provider with no span processors: enough to inspect propagated
-- headers, not enough to inspect exported spans.
setupW3CProvider :: IO OTel.TracerProvider
setupW3CProvider = mkW3CProvider []
-- | A W3C provider whose ended spans are collected in memory. Shut the provider
-- down (see 'capturedSpans') before reading the reference so every span that was
-- still open has been flushed.
setupCapturingProvider :: IO (OTel.TracerProvider, IORef [ImmutableSpan])
setupCapturingProvider = do
(processor, spansRef) <- inMemoryListExporter
provider <- mkW3CProvider [processor]
pure (provider, spansRef)
-- | Shut the provider down (ending and exporting everything still buffered) and
-- return a frozen snapshot of every captured span.
capturedSpans :: OTel.TracerProvider -> IORef [ImmutableSpan] -> IO [CapturedSpan]
capturedSpans provider spansRef = do
_ <- OTel.shutdownTracerProvider provider Nothing
traverse captureSpan =<< readIORef spansRef
-- | A frozen snapshot of an 'ImmutableSpan'. In hs-opentelemetry 1.0 the mutable
-- span fields (name, attributes, status) live behind the @spanHot :: IORef SpanHot@
-- field rather than directly on 'ImmutableSpan', so the tests read that reference
-- once after the span ends and assert on this flat record.
data CapturedSpan = CapturedSpan
{ csName :: Text,
csKind :: OTel.SpanKind,
csAttributes :: Attributes,
csStatus :: OTel.SpanStatus,
csContext :: OTel.SpanContext,
csParent :: Maybe OTel.Span,
csEventNames :: [Text]
}
captureSpan :: ImmutableSpan -> IO CapturedSpan
captureSpan sp = do
hot <- readIORef (spanHot sp)
pure
CapturedSpan
{ csName = hotName hot,
csKind = spanKind sp,
csAttributes = hotAttributes hot,
csStatus = hotStatus hot,
csContext = spanContext sp,
csParent = spanParent sp,
csEventNames =
map eventName (toList (appendOnlyBoundedCollectionValues (hotEvents hot)))
}
textAttr :: Attributes -> Text -> Maybe Text
textAttr attrs name = case lookupAttribute attrs name of
Just (AttributeValue (TextAttribute t)) -> Just t
_ -> Nothing
-- | Every captured span whose name matches exactly.
spansNamed :: Text -> [CapturedSpan] -> [CapturedSpan]
spansNamed name = filter ((== name) . csName)
-- | The 'OTel.SpanContext' of a captured span's parent, if it had one.
parentSpanContext :: CapturedSpan -> IO (Maybe OTel.SpanContext)
parentSpanContext = traverse OTel.getSpanContext . csParent
-- | The one @\<jobName\> process@ span a single one-shot delivery must produce.
-- Anything other than exactly one is a failure that names every captured span, so
-- a duplicate wrapper is diagnosed rather than silently accepted by taking the
-- head of the list.
theProcessSpan :: Text -> [CapturedSpan] -> IO CapturedSpan
theProcessSpan jobName spans =
case spansNamed (jobName <> " process") spans of
[only] -> pure only
other ->
fail
( "expected exactly one "
<> show (jobName <> " process")
<> " span, got "
<> show (length other)
<> "; all captured spans: "
<> show (map csName spans)
)
-- | Run a @Stack@ action against a fresh 'JobRuntime' wired to @tracer@, so both
-- the shibuya 'Tracing' effect and the @pgmq@ interpreter emit spans. Fails the
-- test on any PGMQ runtime error, exactly like 'runDb'.
runDbTraced :: Text -> OTel.Tracer -> Eff Stack a -> IO a
runDbTraced connStr tracer act =
withJobRuntime connStr (Just tracer) $ \rt -> do
res <- runJobEff rt act
either (\e -> fail ("PGMQ runtime error: " <> show e)) pure res
stopAppQuickly :: (IOE :> es) => AppHandle es -> Eff es ()
stopAppQuickly app = do
_ <- stopAppGracefully ShutdownConfig {drainTimeout = 1} app
pure ()
waitUntil :: IO Bool -> IO Bool
waitUntil predicate =
maybe False id <$> timeout 10_000_000 loop
where
loop = do
ok <- predicate
if ok
then pure True
else threadDelay 100_000 >> loop
spec :: SpecWith Text
spec = do
it "round-trips a payload through aesonJobCodec" $ \_connStr -> do
let codec = aesonJobCodec :: JobCodec Ping
sample = Ping "hello" 7
decodeJob codec (encodeJob codec sample) `shouldBe` Right sample
it "round-trips a payload through keiroJobCodec's versioned envelope" $ \_connStr -> do
let codec = keiroJobCodec versionedPingCodec
sample = Ping "hello" 7
decodeJob codec (encodeJob codec sample) `shouldBe` Right sample
it "decodes old keiroJobCodec payloads through the upcaster chain" $ \_connStr -> do
let codec = keiroJobCodec versionedPingCodec
v1Envelope =
object
[ "v" .= (1 :: Int),
"data" .= String "legacy"
]
decodeJob codec v1Envelope `shouldBe` Right (Ping "legacy" 1)
it "classifies future keiroJobCodec payloads as retryable" $ \_connStr -> do
let codec = keiroJobCodec versionedPingCodec
futureEnvelope =
object
[ "v" .= (99 :: Int),
"data" .= object []
]
decodeJob codec futureEnvelope `shouldBe` Left (JobPayloadFromFuture 99 2)
it "classifies malformed keiroJobCodec envelopes as malformed payloads" $ \_connStr -> do
let codec = keiroJobCodec versionedPingCodec
decodeJob codec (String "not an envelope") `shouldSatisfy` \case
Left (JobPayloadMalformed _) -> True
_ -> False
it "validates retry policies" $ \_connStr -> do
mkRetryPolicy 0 (RetryDelay 60) True
`shouldBe` Left (NonPositiveMaxRetries 0)
mkRetryPolicy 1 (RetryDelay (-1)) True
`shouldBe` Left (NegativeRetryDelay (RetryDelay (-1)))
mkRetryPolicy 1 (RetryDelay 0) True
`shouldBe` Right (RetryPolicy 1 (RetryDelay 0) True)
it "validates job tuning" $ \_connStr -> do
mkJobTuning 0 1 (PollEvery 1)
`shouldBe` Left (NonPositiveVisibilityTimeout 0)
mkJobTuning 30 0 (PollEvery 1)
`shouldBe` Left (NonPositiveBatchSize 0)
mkJobTuning 30 1 (PollEvery 0)
`shouldBe` Left NonPositivePollInterval
mkJobTuning 30 1 (LongPoll 0 100)
`shouldBe` Left NonPositivePollInterval
mkJobTuning 30 1 (PollEvery 1)
`shouldBe` Right defaultJobTuning
it "derives distinct physical names for long logical queue names" $ \_connStr -> do
let commonPrefix = Text.replicate 43 "a"
first = queueRef (commonPrefix <> "x")
second = queueRef (commonPrefix <> "y")
first.physicalName `shouldNotBe` second.physicalName
Text.length (queueNameToText first.physicalName) `shouldBe` 43
Text.length (queueNameToText second.physicalName) `shouldBe` 43
it "disambiguates logical names ending in _dlq from derived DLQ names" $ \_connStr -> do
let foo = queueRef "foo"
masquerading = queueRef "foo_dlq"
masqueradingPhysical = queueNameToText masquerading.physicalName
masquerading.physicalName `shouldNotBe` foo.dlqName
masqueradingPhysical `shouldNotSatisfy` Text.isSuffixOf "_dlq"
it "keeps short physical queue names unchanged" $ \_connStr -> do
queueNameToText (queueRef "hospital_capacity.reservation_work").physicalName
`shouldBe` "hospital_capacity_reservation_work"
it "always derives PGMQ-parseable queue names" $ \_connStr -> do
let logicalNames =
[ "",
"!!!",
Text.replicate 100 "x",
"___trailing___",
"foo_dlq",
"9starts.with.digit"
]
traverse_
( \logical -> do
let ref = queueRef logical
physical = queueNameToText ref.physicalName
dlq = queueNameToText ref.dlqName
parseQueueName physical `shouldSatisfy` isRight
parseQueueName dlq `shouldSatisfy` isRight
physical `shouldNotSatisfy` Text.isSuffixOf "_dlq"
)
logicalNames
it "Done deletes the message" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.done"
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "do" 1)
runJobOnce 1 job (\_ -> pure Done)
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
it "Retry redelivers the message" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.retry"
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "again" 2)
runJobOnce 1 job (\_ -> pure (Retry (RetryDelay 0)))
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 1
it "RetryDefault redelivers after the policy default delay" $ \connStr -> do
let job =
(mkJob "keiro_pgmq_test.retry_default")
{ jobPolicy = RetryPolicy 5 (RetryDelay 5) True
}
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "again by default" 2)
runJobOnce 1 job (\_ -> pure RetryDefault)
len <- runDb connStr (queueLen job.jobQueue.physicalName)
emptyImmediateRead <- runDb connStr (readOneIsEmpty job.jobQueue.physicalName)
len `shouldBe` 1
emptyImmediateRead `shouldBe` True
it "runJobOnceWithContext returns promptly when n exceeds the queue length" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.once_short_queue"
result <-
timeout 2_000_000 $
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "only" 1)
runJobOnceWithContext defaultJobTuning 5 job \_ctx _payload ->
pure Done
result `shouldBe` Just 1
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
it "runJobOnceWithContext drains messages in batches greater than one" $ \connStr -> do
handled <- newIORef (0 :: Int)
let job = mkJob "keiro_pgmq_test.once_batch"
tuning =
either (error . show) id $
mkJobTuning 30 2 (PollEvery 1)
drained <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "first" 1)
_ <- enqueue job (Ping "second" 2)
_ <- enqueue job (Ping "third" 3)
runJobOnceWithContext tuning 3 job \_ctx _payload -> do
liftIO $ modifyIORef' handled (+ 1)
pure Done
drained `shouldBe` 3
readIORef handled `shouldReturn` 3
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
it "runJobOnceWithContext Retry delay hides the message until the delay expires" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.once_retry_delay"
drained <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "later" 1)
runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload ->
pure (Retry (RetryDelay 5))
drained `shouldBe` 1
len <- runDb connStr (queueLen job.jobQueue.physicalName)
emptyImmediateRead <- runDb connStr (readOneIsEmpty job.jobQueue.physicalName)
len `shouldBe` 1
emptyImmediateRead `shouldBe` True
it "runJobOnceWithContext leaves thrown-handler messages invisible and continues" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.once_throw"
tuning =
either (error . show) id $
mkJobTuning 2 2 (PollEvery 1)
drained <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "throw" 1)
_ <- enqueue job (Ping "ok" 2)
runJobOnceWithContext tuning 2 job \_ctx payload ->
if payload.message == "throw"
then liftIO $ throwIO (userError "handler failed")
else pure Done
drained `shouldBe` 1
len <- runDb connStr (queueLen job.jobQueue.physicalName)
emptyImmediateRead <- runDb connStr (readOneIsEmpty job.jobQueue.physicalName)
len `shouldBe` 1
emptyImmediateRead `shouldBe` True
threadDelay 2_200_000
drainedAfterVisibilityTimeout <-
runDb connStr $
runJobOnceWithContext tuning 1 job \_ctx _payload ->
pure Done
drainedAfterVisibilityTimeout `shouldBe` 1
it "runJobOnceWithContext auto-routes max-retry messages to the DLQ before rerunning the handler" $ \connStr -> do
callCount <- newIORef (0 :: Int)
let job =
(mkJob "keiro_pgmq_test.once_max_retries")
{ jobPolicy = RetryPolicy 1 (RetryDelay 0) True
}
firstDrain <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "retry-limit" 1)
runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload -> do
liftIO $ modifyIORef' callCount (+ 1)
pure (Retry (RetryDelay 0))
secondDrain <-
runDb connStr $
runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload -> do
liftIO $ modifyIORef' callCount (+ 1)
pure Done
firstDrain `shouldBe` 1
secondDrain `shouldBe` 1
readIORef callCount `shouldReturn` 1
mainLen <- runDb connStr (queueLen job.jobQueue.physicalName)
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
mainLen `shouldBe` 0
dlqLen `shouldBe` 1
it "worker-path lease extension prevents redelivery" $ \connStr -> do
callCount <- newIORef (0 :: Int)
handlerDone <- newIORef False
let job = mkJob "keiro_pgmq_test.worker_lease"
tuning =
either (error . show) id $
mkJobTuning 2 1 (PollEvery 0.2)
processed <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "slow" 4)
result <-
runJobWorkers
IgnoreFailures
16
[ jobProcessorWithContext tuning job \ctx _payload -> do
liftIO $ modifyIORef' callCount (+ 1)
ctx.extendLease 30
liftIO $ threadDelay 4_000_000
liftIO $ writeIORef handlerDone True
pure Done
]
case result of
Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)
Right app -> do
ok <- liftIO $ waitUntil (readIORef handlerDone)
stopAppQuickly app
pure ok
processed `shouldBe` True
readIORef callCount `shouldReturn` 1
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
it "worker-path context exposes the first attempt number" $ \connStr -> do
seenAttempt <- newIORef Nothing
let job = mkJob "keiro_pgmq_test.worker_attempt"
processed <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "attempt" 1)
result <-
runJobWorkers
IgnoreFailures
16
[ jobProcessorWithContext defaultJobTuning job \ctx _payload -> do
liftIO $ writeIORef seenAttempt (Just ctx.attempt)
pure Done
]
case result of
Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)
Right app -> do
ok <- liftIO $ waitUntil ((/= Nothing) <$> readIORef seenAttempt)
stopAppQuickly app
pure ok
processed `shouldBe` True
readIORef seenAttempt `shouldReturn` Just (Just 0)
it "runJobWorkers processes an enqueued message" $ \connStr -> do
processedRef <- newIORef False
let job = mkJob "keiro_pgmq_test.worker_smoke"
processed <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "worker" 1)
result <-
runJobWorkers
IgnoreFailures
16
[ jobProcessor job \_payload -> do
liftIO $ writeIORef processedRef True
pure Done
]
case result of
Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)
Right app -> do
ok <- liftIO $ waitUntil (readIORef processedRef)
stopAppQuickly app
pure ok
processed `shouldBe` True
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
it "runJobWorkers survives a transient database error during polling" $ \_connStr ->
pendingWith "needs a deterministic keiro-pgmq-level transient polling fault injector; EP-1 covers this in upstream shibuya and shibuya-pgmq-adapter tests"
it "worker-path retry limit auto-routes to the DLQ before the handler reruns" $ \connStr -> do
callCount <- newIORef (0 :: Int)
let job =
(mkJob "keiro_pgmq_test.worker_max_retries")
{ jobPolicy = RetryPolicy 1 (RetryDelay 0) True
}
tuning =
either (error . show) id $
mkJobTuning 30 1 (PollEvery 0.1)
dlqReached <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "retry-limit" 1)
result <-
runJobWorkers
IgnoreFailures
16
[ jobProcessorWithContext tuning job \_ctx _payload -> do
liftIO $ modifyIORef' callCount (+ 1)
pure (Retry (RetryDelay 0))
]
case result of
Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)
Right app -> do
ok <- liftIO $ waitUntil do
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
pure (dlqLen == 1)
stopAppQuickly app
pure ok
dlqReached `shouldBe` True
readIORef callCount `shouldReturn` 1
it "enqueueWithDelay delays first delivery" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.enqueue_delay"
runDb connStr $ do
ensureJobQueue job
_ <- enqueueWithDelay job 5 (Ping "later" 1)
pure ()
len <- runDb connStr (queueLen job.jobQueue.physicalName)
emptyImmediateRead <- runDb connStr (readOneIsEmpty job.jobQueue.physicalName)
len `shouldBe` 1
emptyImmediateRead `shouldBe` True
it "Dead routes the message to the DLQ" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dead"
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "poison" 3)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
mainLen <- runDb connStr (queueLen job.jobQueue.physicalName)
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
mainLen `shouldBe` 0
dlqLen `shouldBe` 1
it "readDlq decodes the original dead-lettered payload" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dlq_read"
payload = Ping "poison" 3
entries <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job payload
runJobOnce 1 job (\_ -> pure (Dead "bad"))
readDlq job 1
case entries of
[entry] -> do
entry.reason `shouldSatisfy` Text.isPrefixOf "poison_pill"
entry.originalPayload `shouldBe` Right payload
entry.originalMessageId `shouldSatisfy` (/= Nothing)
entry.readCount `shouldBe` Just 1
_ -> expectationFailure ("expected one DLQ entry, got " <> show (length entries))
it "redriveDlq moves dead-lettered payloads back to the main queue" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dlq_redrive"
redriven <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "redrive" 1)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
redriveDlq job 10
redriven `shouldBe` 1
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
mainLen <- runDb connStr (queueLen job.jobQueue.physicalName)
dlqLen `shouldBe` 0
mainLen `shouldBe` 1
runDb connStr $
runJobOnce 1 job (\_ -> pure Done)
finalMainLen <- runDb connStr (queueLen job.jobQueue.physicalName)
finalMainLen `shouldBe` 0
it "purgeDlq empties the DLQ" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dlq_purge"
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "purge" 1)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
purgeDlq job
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
dlqLen `shouldBe` 0
it "readDlq preserves malformed DLQ wrappers as malformed entries" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dlq_malformed"
entries <-
runDb connStr $ do
ensureJobQueue job
_ <-
Pgmq.sendMessage
SendMessage
{ queueName = job.jobQueue.dlqName,
messageBody = MessageBody (String "not a dlq wrapper"),
delay = Nothing
}
readDlq job 1
case entries of
[entry] -> do
entry.reason `shouldSatisfy` Text.isPrefixOf "malformed_dlq_payload"
entry.originalPayload `shouldSatisfy` \case
Left (JobPayloadMalformed _) -> True
_ -> False
_ -> expectationFailure ("expected one malformed DLQ entry, got " <> show (length entries))
it "undecodable payload routes to the DLQ" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.bad"
runDb connStr $ do
ensureJobQueue job
-- Send raw JSON the Ping codec cannot decode, bypassing enqueue.
_ <-
Pgmq.sendMessage
SendMessage
{ queueName = job.jobQueue.physicalName,
messageBody = MessageBody (String "not a ping"),
delay = Nothing
}
runJobOnce 1 job (\_ -> pure Done)
mainLen <- runDb connStr (queueLen job.jobQueue.physicalName)
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
mainLen `shouldBe` 0
dlqLen `shouldBe` 1
-- EP-1 M1: header-carrying enqueue and the reserved-key contract.
it "enqueueWithHeaders attaches a header readable on the raw PGMQ message" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.hdr_attach"
msgs <-
runDb connStr $ do
ensureJobQueue job
_ <-
enqueueWithHeaders
job
(MessageHeaders (object ["tenant" .= ("acme" :: Text)]))
(Ping "hdr" 1)
readMessages job.jobQueue.physicalName 1
case msgs of
[m] -> headerKey "tenant" m.headers `shouldBe` Just (String "acme")
_ -> expectationFailure ("expected one message, got " <> show (length msgs))
it "enqueueWithHeaders leaves the x-pgmq-group key untouched" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.hdr_group"
msgs <-
runDb connStr $ do
ensureJobQueue job
_ <-
enqueueWithHeaders
job
(MessageHeaders (object ["x-pgmq-group" .= ("g1" :: Text)]))
(Ping "g" 1)
readMessages job.jobQueue.physicalName 1
case msgs of
[m] -> headerKey "x-pgmq-group" m.headers `shouldBe` Just (String "g1")
_ -> expectationFailure ("expected one message, got " <> show (length msgs))
-- EP-1 M2: batch enqueue.
it "enqueueBatch of three payloads yields three ids and queue depth three" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.batch"
ids <-
runDb connStr $ do
ensureJobQueue job
enqueueBatch job [Ping "a" 1, Ping "b" 2, Ping "c" 3]
length ids `shouldBe` 3
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 3
it "enqueueBatchWithHeaders attaches per-message headers" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.batch_headers"
msgs <-
runDb connStr $ do
ensureJobQueue job
_ <-
enqueueBatchWithHeaders
job
[ (MessageHeaders (object ["i" .= (1 :: Int)]), Ping "a" 1),
(MessageHeaders (object ["i" .= (2 :: Int)]), Ping "b" 2)
]
readMessages job.jobQueue.physicalName 2
length msgs `shouldBe` 2
map (headerKey "i" . (.headers)) msgs
`shouldMatchList` [Just (Number 1), Just (Number 2)]
-- EP-1 M3: handler-visible headers and trace propagation.
it "drain-path JobContext exposes the enqueued headers" $ \connStr -> do
seen <- newIORef Nothing
let job = mkJob "keiro_pgmq_test.ctx_headers"
runDb connStr $ do
ensureJobQueue job
_ <-
enqueueWithHeaders
job
(MessageHeaders (object ["tenant" .= ("acme" :: Text)]))
(Ping "h" 1)
_ <-
runJobOnceWithContext defaultJobTuning 1 job \ctx _payload -> do
liftIO (writeIORef seen ctx.headers)
pure Done
pure ()
captured <- readIORef seen
headerKey "tenant" captured `shouldBe` Just (String "acme")
it "a traceparent set at enqueue is visible to the drain-path handler" $ \connStr -> do
seen <- newIORef Nothing
provider <- setupW3CProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.traceparent"
parentSpan <- OTel.createSpan tracer Ctxt.empty "enqueue" OTel.defaultSpanArguments
_ <- CtxtLocal.attachContext (Ctxt.insertSpan parentSpan Ctxt.empty)
runDb connStr $ do
ensureJobQueue job
_ <- enqueueTraced provider job (MessageHeaders (object [])) (Ping "t" 1)
_ <-
runJobOnceWithContext defaultJobTuning 1 job \ctx _payload -> do
liftIO (writeIORef seen ctx.headers)
pure Done
pure ()
OTel.endSpan parentSpan Nothing
captured <- readIORef seen
headerKey "traceparent" captured `shouldSatisfy` \case
Just (String _) -> True
_ -> False
-- EP-111 M1: the captured-span fixture itself, proven against the spans the
-- traced pgmq interpreter already emits.
it "captured tracing fixture sees PGMQ publish and receive spans" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.fixture_spans"
queue = queueNameToText job.jobQueue.physicalName
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <- enqueue job (Ping "fixture" 1)
_ <- readMessages job.jobQueue.physicalName 1
pure ()
spans <- capturedSpans provider spansRef
map csName spans `shouldSatisfy` elem ("publish " <> queue)
map csName spans `shouldSatisfy` elem ("receive " <> queue)
-- EP-111 M2: the central proof — the one-shot process span continues the
-- producer's trace using only what the PGMQ headers carry.
it "one-shot process span continues the enqueued W3C parent" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.one_shot_parent"
producerSpan <- OTel.createSpan tracer Ctxt.empty "enqueue" OTel.defaultSpanArguments
producerCtx <- OTel.getSpanContext producerSpan
-- Attach the producer span only for the enqueue, then detach it. The
-- drain therefore has no local parent to inherit: the only path from
-- producer to consumer is the traceparent stored in the PGMQ headers.
token <- CtxtLocal.attachContext (Ctxt.insertSpan producerSpan Ctxt.empty)
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <- enqueueTraced provider job (MessageHeaders (object [])) (Ping "traced" 1)
pure ()
CtxtLocal.detachContext token
OTel.endSpan producerSpan Nothing
drained <-
runDbTraced connStr tracer $
runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload -> pure Done
drained `shouldBe` 1
spans <- capturedSpans provider spansRef
processSpan <- theProcessSpan job.jobName spans
csKind processSpan `shouldBe` OTel.Consumer
OTel.traceId (csContext processSpan) `shouldBe` OTel.traceId producerCtx
parent <- parentSpanContext processSpan
fmap OTel.spanId parent `shouldBe` Just (OTel.spanId producerCtx)
textAttr (csAttributes processSpan) "messaging.system" `shouldBe` Just "shibuya"
textAttr (csAttributes processSpan) "messaging.destination.name"
`shouldBe` Just job.jobName
textAttr (csAttributes processSpan) "messaging.operation.type" `shouldBe` Just "process"
textAttr (csAttributes processSpan) "messaging.message.id" `shouldSatisfy` \case
Just _ -> True
Nothing -> False
textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Just "ack_ok"
csStatus processSpan `shouldBe` OTel.Ok
-- EP-111 M3: the branches whose telemetry meaning differs from plain success.
it "one-shot Retry reports ack_retry with an OK span and hides the row" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.span_retry"
(drained, len, hidden) <-
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <- enqueue job (Ping "r" 1)
drained <-
runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload ->
pure (Retry (RetryDelay 30))
len <- queueLen job.jobQueue.physicalName
hidden <- readOneIsEmpty job.jobQueue.physicalName
pure (drained, len, hidden)
drained `shouldBe` 1
len `shouldBe` 1
hidden `shouldBe` True
processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef
textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Just "ack_retry"
csStatus processSpan `shouldBe` OTel.Ok
it "one-shot Dead reports ack_dead_letter with an ERROR span" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.span_dead"
(mainLen, dlqLen) <-
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <- enqueue job (Ping "poison" 1)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
mainLen <- queueLen job.jobQueue.physicalName
dlqLen <- queueLen job.jobQueue.dlqName
pure (mainLen, dlqLen)
mainLen `shouldBe` 0
dlqLen `shouldBe` 1
processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef
textAttr (csAttributes processSpan) "shibuya.ack.decision"
`shouldBe` Just "ack_dead_letter"
csStatus processSpan `shouldBe` OTel.Error "poison_pill: bad"
it "an undecodable payload reports ack_dead_letter without a handler-started event" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.span_malformed"
dlqLen <-
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <-
Pgmq.sendMessage
SendMessage
{ queueName = job.jobQueue.physicalName,
messageBody = MessageBody (String "not a ping"),
delay = Nothing
}
runJobOnce 1 job (\_ -> pure Done)
queueLen job.jobQueue.dlqName
dlqLen `shouldBe` 1
processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef
textAttr (csAttributes processSpan) "shibuya.ack.decision"
`shouldBe` Just "ack_dead_letter"
csStatus processSpan `shouldSatisfy` \case
OTel.Error reason -> "invalid_payload: " `Text.isPrefixOf` reason
_ -> False
csEventNames processSpan `shouldNotSatisfy` elem "shibuya.handler.started"
it "a thrown handler records an exception and claims no acknowledgement" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.span_throw"
(drained, len, hidden) <-
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <- enqueue job (Ping "boom" 1)
drained <-
runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload ->
liftIO (throwIO (userError "handler exploded"))
len <- queueLen job.jobQueue.physicalName
hidden <- readOneIsEmpty job.jobQueue.physicalName
pure (drained, len, hidden)
drained `shouldBe` 0
len `shouldBe` 1
hidden `shouldBe` True
processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef
csEventNames processSpan `shouldSatisfy` elem "shibuya.handler.started"
csEventNames processSpan `shouldSatisfy` elem "exception"
csEventNames processSpan `shouldNotSatisfy` elem "shibuya.handler.completed"
textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Nothing
csStatus processSpan `shouldSatisfy` \case
OTel.Error reason -> "handler exception: " `Text.isPrefixOf` reason
_ -> False
it "a message with no trace headers still gets exactly one process span" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.span_no_parent"
-- Plain 'enqueue' writes no headers at all, so there is no traceparent
-- to extract and shibuya falls back to the ambient local context.
runDbTraced connStr tracer $ do
ensureJobQueue job
_ <- enqueue job (Ping "plain" 1)
runJobOnce 1 job (\_ -> pure Done)
processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef
csKind processSpan `shouldBe` OTel.Consumer
textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Just "ack_ok"
csStatus processSpan `shouldBe` OTel.Ok
it "a FIFO delivery carries shibuya.partition on its process span" $ \connStr -> do
(provider, spansRef) <- setupCapturingProvider
let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
job = mkJob "keiro_pgmq_test.span_partition"
drained <-
runDbTraced connStr tracer $ do
ensureOrderedJobQueue job
_ <- enqueueToGroup job "g1" (Ping "grouped" 1)
runJobOnceWithContext (withOrdering FifoThroughput defaultJobTuning) 1 job \_ctx _p ->
pure Done
drained `shouldBe` 1
processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef
textAttr (csAttributes processSpan) "shibuya.partition" `shouldBe` Just "g1"
-- EP-2 M1: unlogged vs standard provisioning.
it "ensureJobQueueWith unlogged creates an unlogged queue" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.unlogged"
unlogged <-
runDb connStr $ do
ensureJobQueueWith unloggedProvision job
queues <- Pgmq.listQueues
pure (queueIsUnlogged job.jobQueue.physicalName queues)
unlogged `shouldBe` Just True
it "ensureJobQueue (standard) creates a logged queue" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.standard_logged"
unlogged <-
runDb connStr $ do
ensureJobQueue job
queues <- Pgmq.listQueues
pure (queueIsUnlogged job.jobQueue.physicalName queues)
unlogged `shouldBe` Just False
-- EP-2 M2: partitioned config shape (pure) + pending live test.
it "ensureJobQueueWith partitioned builds a partitioned QueueConfig" $ \_connStr -> do
let job = mkJob "keiro_pgmq_test.partitioned"
spec = PartitionSpec {partitionInterval = "daily", retentionInterval = "7 days"}
case queueProvisionConfigs (partitionedProvision spec) job of
(mainCfg : _) ->
case mainCfg.queueType of
Config.PartitionedQueue pc -> do
pc.partitionInterval `shouldBe` "daily"
pc.retentionInterval `shouldBe` "7 days"
mainCfg.queueName `shouldBe` job.jobQueue.physicalName
other ->
expectationFailure
("expected PartitionedQueue, got " <> show other)
[] -> expectationFailure "expected at least the main queue config"
it "ensureJobQueueWith partitioned creates a partitioned queue (live)" $ \_connStr ->
pendingWith
"requires a pg_partman-enabled PostgreSQL; the keiro test database installs only \
\the PGMQ schema via pgmq-migration, which does not load pg_partman"
-- EP-2 M3: FIFO index idempotence.
it "ensureFifoIndex is idempotent and the queue still accepts reads" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.fifo_index"
roundTripped <-
runDb connStr $ do
ensureJobQueue job
ensureFifoIndex job
ensureFifoIndex job -- second call must not error
_ <- enqueue job (Ping "after-index" 1)
runJobOnce 1 job (\_ -> pure Done)
queueLen job.jobQueue.physicalName
roundTripped `shouldBe` 0
-- EP-3 M3: group-keyed producer + ordered queue setup.
it "enqueueToGroup writes the x-pgmq-group header" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.group_header"
msgs <-
runDb connStr $ do
ensureOrderedJobQueue job
_ <- enqueueToGroup job "g1" (Ping "grouped" 1)
readMessages job.jobQueue.physicalName 1
case msgs of
[m] -> headerKey "x-pgmq-group" m.headers `shouldBe` Just (String "g1")
_ -> expectationFailure ("expected one message, got " <> show (length msgs))
it "ensureOrderedJobQueue is idempotent and the queue accepts grouped work" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.ordered_setup"
len <-
runDb connStr $ do
ensureOrderedJobQueue job
ensureOrderedJobQueue job -- second call must not error
_ <- enqueueToGroup job "g1" (Ping "x" 1)
_ <-
runJobOnceWithContext (withOrdering FifoThroughput defaultJobTuning) 1 job \_ctx _p ->
pure Done
queueLen job.jobQueue.physicalName
len `shouldBe` 0
-- EP-3 M4: end-to-end ordering proof.
it "FifoThroughput drain preserves strict within-group order and fully drains" $ \connStr -> do
observed <- newIORef ([] :: [Text])
let job = mkJob "keiro_pgmq_test.fifo_order"
drained <-
runDb connStr $ do
ensureOrderedJobQueue job
_ <- enqueueToGroup job "a" (Ping "a1" 1)
_ <- enqueueToGroup job "b" (Ping "b1" 1)
_ <- enqueueToGroup job "a" (Ping "a2" 2)
_ <- enqueueToGroup job "a" (Ping "a3" 3)
_ <- enqueueToGroup job "b" (Ping "b2" 2)
runJobOnceWithContext (withOrdering FifoThroughput defaultJobTuning) 5 job \_ctx payload -> do
liftIO $ modifyIORef' observed (<> [payload.message])
pure Done
log' <- readIORef observed
drained `shouldBe` 5
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
filter (Text.isPrefixOf "a") log' `shouldBe` ["a1", "a2", "a3"]
filter (Text.isPrefixOf "b") log' `shouldBe` ["b1", "b2"]
it "FifoThroughput worker path preserves within-group order" $ \connStr -> do
observed <- newIORef ([] :: [Text])
let job = mkJob "keiro_pgmq_test.fifo_worker"
tuning =
withOrdering FifoThroughput $
either (error . show) id $
mkJobTuning 30 1 (PollEvery 0.1)
processed <-
runDb connStr $ do
ensureOrderedJobQueue job
_ <- enqueueToGroup job "a" (Ping "a1" 1)
_ <- enqueueToGroup job "a" (Ping "a2" 2)
_ <- enqueueToGroup job "a" (Ping "a3" 3)
result <-
runJobWorkers
IgnoreFailures
16
[ jobProcessorWithContext tuning job \_ctx payload -> do
liftIO $ modifyIORef' observed (<> [payload.message])
pure Done
]
case result of
Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)
Right app -> do
ok <- liftIO $ waitUntil ((>= 3) . length <$> readIORef observed)
stopAppQuickly app
pure ok
processed `shouldBe` True
log' <- readIORef observed
filter (Text.isPrefixOf "a") log' `shouldBe` ["a1", "a2", "a3"]
len <- runDb connStr (queueLen job.jobQueue.physicalName)
len `shouldBe` 0
-- EP-4 M1: typed metrics surface (main + DLQ).
it "jobQueueMetrics reports main-queue depth after enqueue" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.metrics_depth"
(mainMetrics, dlqMetrics) <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "a" 1)
_ <- enqueue job (Ping "b" 2)
_ <- enqueue job (Ping "c" 3)
mainMetrics <- jobQueueMetrics job
dlqMetrics <- jobDlqMetrics job
pure (mainMetrics, dlqMetrics)
mainMetrics.queueLength `shouldBe` 3
mainMetrics.queueVisibleLength `shouldBe` 3
dlqMetrics.queueLength `shouldBe` 0
it "jobDlqMetrics reports DLQ depth after a Dead outcome" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.metrics_dlq"
(mainMetrics, dlqMetrics) <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "poison" 1)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
mainMetrics <- jobQueueMetrics job
dlqMetrics <- jobDlqMetrics job
pure (mainMetrics, dlqMetrics)
mainMetrics.queueLength `shouldBe` 0
dlqMetrics.queueLength `shouldBe` 1
-- EP-4 M2: archive/retention API.
it "archiveDlq retains dead-lettered rows in the archive table" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dlq_archive"
(archived, dlqLen) <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "poison" 1)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
archived <- archiveDlq job 10
dlqMetrics <- jobDlqMetrics job
pure (archived, dlqMetrics.queueLength)
archived `shouldBe` 1
dlqLen `shouldBe` 0
retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)
retained `shouldBe` 1
-- EP-4 M3: end-to-end retention lifecycle.
it "archived DLQ rows survive a purge" $ \connStr -> do
let job = mkJob "keiro_pgmq_test.dlq_archive_purge"
archived <-
runDb connStr $ do
ensureJobQueue job
_ <- enqueue job (Ping "poison" 1)
runJobOnce 1 job (\_ -> pure (Dead "bad"))
archived <- archiveDlq job 10
purgeDlq job
pure archived
archived `shouldBe` 1
dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
dlqLen `shouldBe` 0
retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)
retained `shouldBe` 1