keiro-0.22.0.0: src/Keiro/Timer/Schema.hs
-- | The @keiro_timers@ table: storage and claim logic for durable timers.
--
-- Holds one row per scheduled timer with its 'TimerStatus' lifecycle.
-- 'scheduleTimerTx' inserts (or re-arms a still-@Scheduled@ timer with the same
-- id) inside the caller's transaction; 'claimDueTimer' atomically picks the
-- single earliest due timer with @FOR UPDATE SKIP LOCKED@ and moves it to
-- @Firing@, so competing workers never claim the same timer; 'markTimerFired'
-- records completion and the produced event id. Stale @Firing@ rows are requeued
-- by 'requeueStuckTimers' so a crashed worker does not strand a timer forever.
--
-- ID-only completion, cancellation, dead-lettering and requeueing refuse all
-- token-bearing foreground claims, including expired claims awaiting recovery.
--
-- Callers normally use the re-exports from "Keiro.Timer" rather than this
-- module directly.
module Keiro.Timer.Schema
( -- * Rows and status
TimerStatus (..),
TimerRow (..),
-- * Read-only inspection
TimerInspection (..),
TimerReasonFilter (..),
DeadTimerFilter (..),
anyDeadTimer,
DeadTimerPageRequest (..),
DeadTimerReadError (..),
DeadTimerPage (..),
lookupTimerInspection,
findDeadTimers,
PendingTimerFilter (..),
anyPendingTimer,
PendingTimerPageRequest (..),
PendingTimerReadError (..),
PendingTimerPage (..),
pendingTimerStatuses,
pendingTimerCursorKind,
validatePendingTimerRequest,
findPendingTimers,
pendingTimerItemJson,
pendingTimerPageJson,
-- * Guarded foreground resume
DeadTimerClaimRequest (..),
TimerResumeError (..),
TimerResumeClaim,
resumeClaimTimer,
resumeClaimLeaseUntil,
claimDeadTimer,
renewTimerResume,
completeTimerResume,
parkTimerResume,
cancelTimerResume,
recoverExpiredTimerResumes,
-- * Storage
scheduleTimerTx,
scheduleTimerOnceTx,
claimDueTimer,
lookupTimer,
markTimerFired,
-- * Read-only counts
countDueTimers,
countStuckTimers,
-- * Recovery
StuckTimerFilter (..),
anyStuckTimer,
findStuckTimers,
requeueStuckTimers,
requeueStuckTimer,
cancelTimerTx,
cancelTimer,
deadLetterTimer,
)
where
import Contravariant.Extras (contrazip2, contrazip5, contrazip6)
import Data.Aeson (object, (.=))
import Data.Int (Int32)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Time (NominalDiffTime, addUTCTime, diffUTCTime)
import Data.UUID (UUID)
import Data.UUID qualified as UUID
import Data.UUID.V4 qualified as UUIDv4
import Effectful (Eff, IOE, (:>))
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Statement (Statement, preparable)
import Keiro.Inspection.Cursor
import Keiro.Prelude hiding ((.=))
import Keiro.Timer.Types (TimerId (..), TimerRequest (..))
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Transaction (runTransaction)
import Kiroku.Store.Types (EventId (..))
import "hasql-transaction" Hasql.Transaction qualified as Tx
-- | A timer's lifecycle state.
--
-- * 'Scheduled' — waiting for its 'fireAt'; claimable.
-- * 'Firing' — claimed by a worker and being processed; stale rows become
-- claimable again when 'requeueStuckTimers' moves them back to 'Scheduled'.
-- * 'Fired' — successfully fired; terminal.
-- * 'Cancelled' — withdrawn before firing.
-- * 'Dead' — parked or abandoned; guarded foreground resume is possible; carries an
-- optional @last_error@ describing why it was given up on.
data TimerStatus
= Scheduled
| Firing
| Fired
| Cancelled
| Dead
-- 'Enum'/'Bounded' let a consumer enumerate the complete lifecycle rather than
-- restate it; keiro-dsl's cross-package vocabulary test relies on this.
deriving stock (Generic, Eq, Show, Enum, Bounded)
-- | A timer row as stored: the original 'TimerRequest' fields plus the live
-- 'status', the 'attempts' count (incremented on each claim), and the
-- 'firedEventId' recorded once it fires.
data TimerRow = TimerRow
{ timerId :: !TimerId,
processManagerName :: !Text,
correlationId :: !Text,
fireAt :: !UTCTime,
payload :: !Value,
status :: !TimerStatus,
attempts :: !Int,
firedEventId :: !(Maybe EventId)
}
deriving stock (Generic, Eq, Show)
-- | Original timer metadata and the full stored reason. NULL and empty text
-- remain distinct. Reading does not claim work or authorize its disclosure.
data TimerInspection = TimerInspection
{ timer :: !TimerRow,
lastError :: !(Maybe Text)
}
deriving stock (Generic, Eq, Show)
-- | Case-sensitive literal reason matching. An empty prefix matches every
-- non-NULL reason; percent, underscore, and backslash are ordinary characters.
data TimerReasonFilter
= AnyTimerReason
| ReasonAbsent
| ReasonExact !Text
| ReasonPrefix !Text
deriving stock (Generic, Eq, Show)
-- | Dead rows matching both the optional exact owner and the reason predicate.
-- The owner label is not an application authorization credential.
data DeadTimerFilter = DeadTimerFilter
{ processManagerName :: !(Maybe Text),
reason :: !TimerReasonFilter
}
deriving stock (Generic, Eq, Show)
-- | Select all dead timers.
anyDeadTimer :: DeadTimerFilter
anyDeadTimer = DeadTimerFilter Nothing AnyTimerReason
-- | Request 1 through 100 rows, strictly after an optional UUID cursor.
-- Restart without a cursor when changing filters.
data DeadTimerPageRequest = DeadTimerPageRequest
{ pageSize :: !Int,
afterTimerId :: !(Maybe TimerId)
}
deriving stock (Generic, Eq, Show)
-- | Invalid sizes are rejected before database access, without clamping.
data DeadTimerReadError = InvalidDeadTimerPageSize !Int
deriving stock (Generic, Eq, Show)
-- | Ascending UUID order, not chronological order. Continuation exists only
-- when another matching row was observed. Requests see current eligibility,
-- not a shared snapshot: newly eligible IDs behind the cursor are not revisited.
data DeadTimerPage = DeadTimerPage
{ timers :: ![TimerInspection],
nextAfterTimerId :: !(Maybe TimerId)
}
deriving stock (Generic, Eq, Show)
-- | Inspect any lifecycle state without mutations or row-claim locks.
lookupTimerInspection :: (Store :> es) => TimerId -> Eff es (Maybe TimerInspection)
lookupTimerInspection timerId =
runTransaction $ Tx.statement (timerIdToUuid timerId) lookupTimerInspectionStmt
-- | Observe a bounded page of dead timers. Limits bound returned rows, not
-- database search cost. Callers must decode and authorize before rendering,
-- following storage continuation even if authorization removes an entire page.
findDeadTimers ::
(Store :> es) =>
DeadTimerFilter ->
DeadTimerPageRequest ->
Eff es (Either DeadTimerReadError DeadTimerPage)
findDeadTimers deadFilter request
| size < 1 || size > 100 = pure (Left (InvalidDeadTimerPageSize size))
| otherwise = do
rows <-
runTransaction $
Tx.statement
( deadFilter ^. #processManagerName,
mode,
reasonText,
timerIdToUuid <$> request ^. #afterTimerId,
fromIntegral size + 1
)
findDeadTimersStmt
let selected = take size rows
continuation = case drop size rows of
[] -> Nothing
_ -> case reverse selected of
lastRow : _ -> Just (lastRow ^. #timer . #timerId)
[] -> Nothing
pure (Right (DeadTimerPage selected continuation))
where
size = request ^. #pageSize
(mode, reasonText) = case deadFilter ^. #reason of
AnyTimerReason -> (0, Nothing)
ReasonAbsent -> (1, Nothing)
ReasonExact value -> (2, Just value)
ReasonPrefix value -> (3, Just value)
-- | Exact optional owner/correlation filters. These reads reserve no work.
data PendingTimerFilter = PendingTimerFilter
{ processManagerName :: !(Maybe Text),
correlationId :: !(Maybe Text)
}
deriving stock (Generic, Eq, Show)
anyPendingTimer :: PendingTimerFilter
anyPendingTimer = PendingTimerFilter Nothing Nothing
data PendingTimerPageRequest = PendingTimerPageRequest
{ pageSize :: !Int,
after :: !(Maybe InspectionCursor)
}
deriving stock (Generic, Eq, Show)
data PendingTimerReadError
= InvalidPendingTimerPageSize !Int
| InvalidPendingTimerCursor !CursorError
| PendingTimerResultTooLarge
deriving stock (Generic, Eq, Show)
-- | Live eligibility, ordered by (fire_at,timer_id), not a frozen traversal.
-- Limits bound returned rows, not search work. In particular filtered reads
-- can examine the pending ledger; automatic callers must apply their budget.
data PendingTimerPage = PendingTimerPage
{ observedAt :: !UTCTime,
timers :: ![TimerInspection],
next :: !(Maybe InspectionCursor)
}
deriving stock (Generic, Eq, Show)
pendingTimerStatuses :: NonEmpty TimerStatus
pendingTimerStatuses = Scheduled :| [Firing]
pendingTimerCursorKind :: Text
pendingTimerCursorKind = "timer-pending"
-- | Pure validation, before borrowing a connection or hydrating an instance.
-- Continuations bind both filters as well as the exclusive database key.
validatePendingTimerRequest :: PendingTimerFilter -> PendingTimerPageRequest -> Either PendingTimerReadError (Maybe (UTCTime, UUID))
validatePendingTimerRequest timerFilter request
| request ^. #pageSize < 1 || request ^. #pageSize > 500 = Left (InvalidPendingTimerPageSize (request ^. #pageSize))
| otherwise = traverse decode (request ^. #after)
where
invalid = Left . InvalidPendingTimerCursor
decode cursor = case decodeCursor pendingTimerCursorKind cursor of
Left err -> invalid err
Right (manager, correlation, fireAt, timerText)
| manager /= timerFilter ^. #processManagerName || correlation /= timerFilter ^. #correlationId -> invalid (MalformedCursor "cursor belongs to another timer filter")
| otherwise -> case UUID.fromText timerText of
Nothing -> invalid (MalformedCursor "invalid timer UUID")
Just timerId -> Right (fireAt, timerId)
-- | Read pending timers and database time in one observational transaction.
-- No locks, claims, cleanup or mutation. Statement shapes avoid nullable OR
-- filters so custom and generic plans can use the available ordered indexes.
findPendingTimers :: (Store :> es) => PendingTimerFilter -> PendingTimerPageRequest -> Eff es (Either PendingTimerReadError PendingTimerPage)
findPendingTimers timerFilter request = case validatePendingTimerRequest timerFilter request of
Left err -> pure (Left err)
Right position -> do
(observed, rows) <- runTransaction $ do
observed <- Tx.statement () (preparable "SELECT now()" mempty (D.singleRow (D.column (D.nonNullable D.timestamptz))))
rows <-
Tx.statement
(timerFilter ^. #processManagerName, timerFilter ^. #correlationId, fst <$> position, snd <$> position, fromIntegral size + 1)
(findPendingTimersStmt (isJust (timerFilter ^. #processManagerName)) (isJust (timerFilter ^. #correlationId)) (isJust position))
pure (observed, rows)
let selected = take size rows
continuation = case (drop size rows, reverse selected) of
(_ : _, lastRow : _) ->
let timer = lastRow ^. #timer
in Just
<$> maybe
(Left PendingTimerResultTooLarge)
Right
( encodeCursorBounded
pendingTimerCursorKind
(timerFilter ^. #processManagerName, timerFilter ^. #correlationId, timer ^. #fireAt, UUID.toText (timerIdToUuid (timer ^. #timerId)))
)
_ -> Right Nothing
pure (PendingTimerPage observed selected <$> continuation)
where
size = request ^. #pageSize
-- | Shared representation for pending pages and process-manager details.
pendingTimerItemJson :: UTCTime -> TimerInspection -> Value
pendingTimerItemJson observed inspection =
let timer = inspection ^. #timer
in object
[ "timer_id" .= UUID.toText (timerIdToUuid (timer ^. #timerId)),
"process_manager_name" .= (timer ^. #processManagerName),
"correlation_id" .= (timer ^. #correlationId),
"fire_at" .= (timer ^. #fireAt),
"payload" .= (timer ^. #payload),
"status" .= statusToText (timer ^. #status),
"attempts" .= (timer ^. #attempts),
"fired_event_id" .= fmap (UUID.toText . eventIdToUuid) (timer ^. #firedEventId),
"last_error" .= (inspection ^. #lastError),
"due_in_seconds" .= (realToFrac (diffUTCTime (timer ^. #fireAt) observed) :: Double)
]
pendingTimerPageJson :: PendingTimerPage -> Value
pendingTimerPageJson page =
object
( ["observed_at" .= (page ^. #observedAt), "items" .= map (pendingTimerItemJson (page ^. #observedAt)) (page ^. #timers)]
<> maybe [] (\cursor -> ["next_cursor" .= cursorText cursor]) (page ^. #next)
)
findPendingTimersStmt :: Bool -> Bool -> Bool -> Statement (Maybe Text, Maybe Text, Maybe UTCTime, Maybe UUID, Int64) [TimerInspection]
findPendingTimersStmt manager correlation cursor =
preparable
( "SELECT timer_id,process_manager_name,correlation_id,fire_at,payload,status,attempts,fired_event_id,last_error FROM keiro.keiro_timers WHERE status IN ('scheduled','firing')"
<> (if manager then " AND process_manager_name=$1" else " AND $1::text IS NULL")
<> (if correlation then " AND correlation_id=$2" else " AND $2::text IS NULL")
<> (if cursor then " AND (fire_at,timer_id)>($3,$4)" else " AND $3::timestamptz IS NULL AND $4::uuid IS NULL")
<> " ORDER BY fire_at,timer_id LIMIT $5"
)
( contrazip5
(E.param (E.nullable E.text))
(E.param (E.nullable E.text))
(E.param (E.nullable E.timestamptz))
(E.param (E.nullable E.uuid))
(E.param (E.nonNullable E.int8))
)
(D.rowList timerInspectionDecoder)
-- | Exact dead-row guards and an explicit total attempt ceiling. Lease seconds
-- must be between 1 and 2147483647, avoiding interval conversion overflow.
data DeadTimerClaimRequest = DeadTimerClaimRequest
{ timerId :: !TimerId,
processManagerName :: !Text,
expectedReason :: !Text,
maxAttempts :: !Int,
leaseSeconds :: !Int
}
deriving stock (Generic, Eq, Show)
data TimerResumeError
= InvalidTimerResumeMaxAttempts !Int
| InvalidTimerResumeLeaseSeconds !Int
deriving stock (Generic, Eq, Show)
-- | Opaque storage ownership, not application authorization.
data TimerResumeClaim = TimerResumeClaim !TimerRow !UUID !UTCTime
-- | Original work as claimed, including the incremented attempt count.
resumeClaimTimer :: TimerResumeClaim -> TimerRow
resumeClaimTimer (TimerResumeClaim row _ _) = row
-- | Claim-time snapshot only. Renewals retain the token and update the database;
-- schedule renewals by the requested interval, not this old snapshot.
resumeClaimLeaseUntil :: TimerResumeClaim -> UTCTime
resumeClaimLeaseUntil (TimerResumeClaim _ _ deadline) = deadline
validResumeLease :: Int -> Bool
validResumeLease seconds = seconds > 0 && toInteger seconds <= toInteger (maxBound :: Int32)
-- | Claim only the exact owner and non-NULL reason. Refusal consumes no attempt.
-- Authorize and establish session availability before calling; execute outside
-- the retried SQL transaction, only after receiving ownership.
claimDeadTimer :: (IOE :> es, Store :> es) => DeadTimerClaimRequest -> Eff es (Either TimerResumeError (Maybe TimerResumeClaim))
claimDeadTimer request
| request ^. #maxAttempts < 0 = pure (Left (InvalidTimerResumeMaxAttempts (request ^. #maxAttempts)))
| not (validResumeLease (request ^. #leaseSeconds)) = pure (Left (InvalidTimerResumeLeaseSeconds (request ^. #leaseSeconds)))
| otherwise = do
token <- liftIO UUIDv4.nextRandom
Right
<$> runTransaction
( do
locked <- lockTimerResumeTx (request ^. #timerId)
if not locked
then pure Nothing
else
Tx.statement
( timerIdToUuid (request ^. #timerId),
request ^. #processManagerName,
request ^. #expectedReason,
fromIntegral (request ^. #maxAttempts),
token,
fromIntegral (request ^. #leaseSeconds)
)
claimDeadTimerStmt
)
-- Lock in a separate statement: subsequent predicates and clock_timestamp()
-- observe the committed winner after a ReadCommitted lock wait. A missing row
-- must return immediately: a later insert must not bypass the initial lock.
lockTimerResumeTx :: TimerId -> Tx.Transaction Bool
lockTimerResumeTx tid = isJust <$> Tx.statement (timerIdToUuid tid) lockTimerResumeStmt
lockTimerResumeStmt :: Statement UUID (Maybe UUID)
lockTimerResumeStmt =
preparable
"SELECT timer_id FROM keiro.keiro_timers WHERE timer_id = $1 FOR UPDATE"
(E.param (E.nonNullable E.uuid))
(D.rowMaybe (D.column (D.nonNullable D.uuid)))
claimDeadTimerStmt :: Statement (UUID, Text, Text, Int64, UUID, Int32) (Maybe TimerResumeClaim)
claimDeadTimerStmt =
preparable
"""
WITH stamp AS MATERIALIZED (SELECT clock_timestamp() AS now)
UPDATE keiro.keiro_timers kt
SET status = 'firing', attempts = attempts + 1,
resume_claim_token = $5, resume_lease_until = stamp.now + $6::integer * interval '1 second',
updated_at = stamp.now
FROM stamp
WHERE timer_id = $1 AND status = 'dead'
AND process_manager_name COLLATE "C" = $2
AND last_error COLLATE "C" = $3 AND attempts < $4
RETURNING kt.timer_id, kt.process_manager_name, kt.correlation_id, kt.fire_at,
kt.payload, kt.status, kt.attempts, kt.fired_event_id, kt.resume_claim_token, kt.resume_lease_until
"""
( contrazip6
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.int8))
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.int4))
)
(D.rowMaybe (TimerResumeClaim <$> timerRowDecoder <*> D.column (D.nonNullable D.uuid) <*> D.column (D.nonNullable D.timestamptz)))
-- | Extend from database time. False means ownership was lost; an expired claim
-- cannot be revived. Stop local work where possible on loss of ownership.
renewTimerResume :: (Store :> es) => TimerResumeClaim -> Int -> Eff es (Either TimerResumeError Bool)
renewTimerResume claim seconds
| not (validResumeLease seconds) = pure (Left (InvalidTimerResumeLeaseSeconds seconds))
| otherwise = Right <$> mutateTimerResume claim "firing" Nothing (Just (fromIntegral seconds))
-- | Complete with the resulting event. False must not be reported as successful
-- timer completion. External effects still need caller-owned idempotency.
completeTimerResume :: (Store :> es) => TimerResumeClaim -> EventId -> Eff es Bool
completeTimerResume claim event = mutateTimerResume claim "fired" (Just (eventIdToUuid event)) Nothing
-- | Return to Dead with the original reason and incremented attempts retained.
parkTimerResume :: (Store :> es) => TimerResumeClaim -> Eff es Bool
parkTimerResume claim = mutateTimerResume claim "dead" Nothing Nothing
-- | Explicit abandonment by the current owner.
cancelTimerResume :: (Store :> es) => TimerResumeClaim -> Eff es Bool
cancelTimerResume claim = mutateTimerResume claim "cancelled" Nothing Nothing
mutateTimerResume :: (Store :> es) => TimerResumeClaim -> Text -> Maybe UUID -> Maybe Int32 -> Eff es Bool
mutateTimerResume (TimerResumeClaim row token _) target event seconds = runTransaction $ do
locked <- lockTimerResumeTx (row ^. #timerId)
if not locked
then pure False
else
Tx.statement (timerIdToUuid (row ^. #timerId), token, target, event, seconds) mutateTimerResumeStmt
mutateTimerResumeStmt :: Statement (UUID, UUID, Text, Maybe UUID, Maybe Int32) Bool
mutateTimerResumeStmt =
preparable
"""
WITH stamp AS MATERIALIZED (SELECT clock_timestamp() AS now)
UPDATE keiro.keiro_timers
SET status = $3,
fired_event_id = CASE WHEN $3 = 'fired' THEN $4 ELSE fired_event_id END,
resume_claim_token = CASE WHEN $5::integer IS NULL THEN NULL ELSE resume_claim_token END,
resume_lease_until = CASE WHEN $5::integer IS NULL THEN NULL ELSE stamp.now + $5 * interval '1 second' END,
updated_at = stamp.now
FROM stamp
WHERE timer_id = $1 AND resume_claim_token = $2 AND status = 'firing'
AND resume_lease_until > stamp.now
"""
( contrazip5
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.text))
(E.param (E.nullable E.uuid))
(E.param (E.nullable E.int4))
)
((> 0) <$> D.rowsAffected)
-- | Re-park expired foreground work directly to Dead. Foreground-only hosts
-- must run this periodically and before discovery/resume. Ordinary worker passes
-- also run it, independently of their ordinary stale-claim recovery option.
recoverExpiredTimerResumes :: (Store :> es) => Eff es Int
recoverExpiredTimerResumes = runTransaction $ do
-- Deterministic lock ordering, with a fresh predicate in the second statement.
ids <- Tx.statement () lockExpiredTimerResumesStmt
sum <$> traverse (\tid -> Tx.statement tid recoverExpiredTimerResumesStmt) ids
lockExpiredTimerResumesStmt :: Statement () [UUID]
lockExpiredTimerResumesStmt =
preparable
"""
SELECT timer_id FROM keiro.keiro_timers
WHERE status = 'firing' AND resume_claim_token IS NOT NULL
AND resume_lease_until <= clock_timestamp()
ORDER BY timer_id FOR UPDATE
"""
mempty
(D.rowList (D.column (D.nonNullable D.uuid)))
recoverExpiredTimerResumesStmt :: Statement UUID Int
recoverExpiredTimerResumesStmt =
preparable
"""
UPDATE keiro.keiro_timers
SET status = 'dead', resume_claim_token = NULL, resume_lease_until = NULL,
updated_at = clock_timestamp()
WHERE timer_id = $1 AND status = 'firing' AND resume_claim_token IS NOT NULL
AND resume_lease_until <= clock_timestamp()
"""
(E.param (E.nonNullable E.uuid))
(fromIntegral <$> D.rowsAffected)
-- | Criteria selecting timers stranded in 'Firing'. A row is "stuck" when its
-- 'status' is @firing@ and it matches every set bound: 'minAge' (it has been
-- firing at least this long, measured from @updated_at@) and 'minAttempts' (it
-- has been claimed at least this many times). Both unset selects every @firing@
-- row.
data StuckTimerFilter = StuckTimerFilter
{ minAge :: !(Maybe NominalDiffTime),
minAttempts :: !(Maybe Int)
}
deriving stock (Generic, Eq, Show)
-- | Select every @firing@ row regardless of age or attempts.
anyStuckTimer :: StuckTimerFilter
anyStuckTimer = StuckTimerFilter Nothing Nothing
-- | Schedule a timer inside the caller's transaction (typically a process
-- manager's append). Upserts on 'timerId': a conflicting row is re-armed only
-- while it is still @Scheduled@, so a timer that has already fired or been
-- cancelled is not resurrected.
scheduleTimerTx :: TimerRequest -> Tx.Transaction ()
scheduleTimerTx request =
Tx.statement
( timerIdToUuid (request ^. #timerId),
request ^. #processManagerName,
request ^. #correlationId,
request ^. #fireAt,
request ^. #payload,
statusToText Scheduled
)
scheduleTimerStmt
-- | Schedule a timer only if no row with the same 'timerId' already exists.
--
-- This is for callers whose first arm must win, such as durable workflow sleeps:
-- every resume pass re-runs the sleep arm until the timer fires, and preserving
-- the original 'fireAt' keeps the sleep measured from the first arm. Process
-- managers that intentionally push a deadline back should keep using
-- 'scheduleTimerTx'. Returns 'True' when this call inserted the row and 'False'
-- when an existing timer won.
scheduleTimerOnceTx :: TimerRequest -> Tx.Transaction Bool
scheduleTimerOnceTx request =
Tx.statement
( timerIdToUuid (request ^. #timerId),
request ^. #processManagerName,
request ^. #correlationId,
request ^. #fireAt,
request ^. #payload,
statusToText Scheduled
)
scheduleTimerOnceStmt
-- | Atomically claim the single earliest timer due at @now@, moving it to
-- @Firing@ and bumping its attempt count. Uses @FOR UPDATE SKIP LOCKED@ so
-- concurrent workers each get a distinct timer. Returns 'Nothing' when none is
-- due.
claimDueTimer :: (Store :> es) => UTCTime -> Eff es (Maybe TimerRow)
claimDueTimer now =
runTransaction $
Tx.statement now claimDueTimerStmt
-- | Look up one timer by its stable identifier without claiming or mutating it.
--
-- Operational tooling uses this to render an exact preview before invoking one
-- of the guarded lifecycle transitions below.
lookupTimer :: (Store :> es) => TimerId -> Eff es (Maybe TimerRow)
lookupTimer timerId =
runTransaction $
Tx.statement (timerIdToUuid timerId) lookupTimerStmt
-- | Mark a claimed timer @Fired@, recording the id of the event its firing
-- produced. Returns 'False' when the row left @Firing@ while the fire action was
-- running (for example, it was requeued, cancelled, or dead-lettered).
markTimerFired :: (Store :> es) => TimerId -> EventId -> Eff es Bool
markTimerFired timerId eventId =
runTransaction $
Tx.statement (timerIdToUuid timerId, eventIdToUuid eventId) markTimerFiredStmt
-- | Count timers that are @scheduled@ and already due at @now@ — the timer
-- backlog. Read-only; mirrors 'claimDueTimer''s WHERE clause but counts rather
-- than locking, so it never claims or mutates a row.
countDueTimers :: (Store :> es) => UTCTime -> Eff es Int
countDueTimers now =
runTransaction $
Tx.statement now countDueTimersStmt
-- | Count timers stranded in @Firing@ that match the given 'StuckTimerFilter' —
-- the same "stuck" predicate 'findStuckTimers' lists, evaluated against @now@.
-- Read-only. 'anyStuckTimer' counts every @firing@ row.
countStuckTimers :: (Store :> es) => UTCTime -> StuckTimerFilter -> Eff es Int
countStuckTimers now stuckFilter =
runTransaction $
Tx.statement (cutoff, fmap fromIntegral (stuckFilter ^. #minAttempts)) countStuckTimersStmt
where
cutoff = fmap (\age -> addUTCTime (negate age) now) (stuckFilter ^. #minAge)
-- | List timers stranded in @Firing@ that match the given 'StuckTimerFilter'.
-- The @minAge@ bound is evaluated against @now@: a row qualifies when its
-- @updated_at@ is at least @minAge@ in the past (cutoff @now - minAge@). Results
-- are ordered oldest-first by @updated_at@.
findStuckTimers ::
(Store :> es) => UTCTime -> StuckTimerFilter -> Eff es [TimerRow]
findStuckTimers now stuckFilter =
runTransaction $
Tx.statement (cutoff, fmap fromIntegral (stuckFilter ^. #minAttempts)) findStuckTimersStmt
where
cutoff = fmap (\age -> addUTCTime (negate age) now) (stuckFilter ^. #minAge)
-- | Move every timer stranded in @Firing@ for at least @olderThan@ back to
-- @Scheduled@. The statement preserves @fire_at@, so a due timer becomes
-- claimable on the same worker pass. Returns the number of rows requeued.
requeueStuckTimers :: (Store :> es) => NominalDiffTime -> UTCTime -> Eff es Int
requeueStuckTimers olderThan now =
runTransaction $
Tx.statement cutoff requeueStuckTimersStmt
where
cutoff = addUTCTime (negate olderThan) now
-- | Move a timer from @Firing@ back to @Scheduled@ so the ordinary claim loop
-- re-fires it. Leaves @fire_at@ unchanged, so a due timer becomes immediately
-- re-claimable. Idempotent: only @firing@ rows match, so re-running on an
-- already-requeued row affects nothing. Returns 'True' when a row changed.
requeueStuckTimer :: (Store :> es) => TimerId -> Eff es Bool
requeueStuckTimer timerId =
runTransaction $
Tx.statement (timerIdToUuid timerId) requeueStuckTimerStmt
-- | Move a timer from @Scheduled@ or @Firing@ to the terminal @Cancelled@
-- state so it never fires. Terminal rows (@fired@, @cancelled@, @dead@) are left
-- untouched. Idempotent. Returns 'True' when a row changed.
cancelTimer :: (Store :> es) => TimerId -> Eff es Bool
cancelTimer = runTransaction . cancelTimerTx
-- | Transactional form of 'cancelTimer'. It uses the same guarded SQL but lets
-- callers compose cancellation atomically with an event append and other timer
-- mutations. Foreground-owned rows remain protected even after their lease has
-- expired; recovery must clear their ownership token first.
cancelTimerTx :: TimerId -> Tx.Transaction Bool
cancelTimerTx timerId =
Tx.statement (timerIdToUuid timerId) cancelTimerStmt
-- | Move a timer from @Scheduled@ or @Firing@ to the terminal @Dead@ state,
-- recording @reason@ in @last_error@ so an operator can see why it was abandoned
-- through 'lookupTimerInspection' or 'findDeadTimers'. Terminal rows are left
-- untouched. Idempotent. Returns 'True' when a row changed.
deadLetterTimer :: (Store :> es) => TimerId -> Text -> Eff es Bool
deadLetterTimer timerId reason =
runTransaction $
Tx.statement (timerIdToUuid timerId, reason) deadLetterTimerStmt
scheduleTimerStmt :: Statement (UUID, Text, Text, UTCTime, Value, Text) ()
scheduleTimerStmt =
preparable
"""
INSERT INTO keiro.keiro_timers
(timer_id, process_manager_name, correlation_id, fire_at, payload, status)
VALUES
($1, $2, $3, $4, $5, $6)
ON CONFLICT (timer_id) DO UPDATE
SET process_manager_name = EXCLUDED.process_manager_name,
correlation_id = EXCLUDED.correlation_id,
fire_at = EXCLUDED.fire_at,
payload = EXCLUDED.payload,
status = EXCLUDED.status,
updated_at = now()
WHERE keiro_timers.status = 'scheduled'
"""
( contrazip6
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.timestamptz))
(E.param (E.nonNullable E.jsonb))
(E.param (E.nonNullable E.text))
)
D.noResult
scheduleTimerOnceStmt :: Statement (UUID, Text, Text, UTCTime, Value, Text) Bool
scheduleTimerOnceStmt =
preparable
"""
INSERT INTO keiro.keiro_timers
(timer_id, process_manager_name, correlation_id, fire_at, payload, status)
VALUES
($1, $2, $3, $4, $5, $6)
ON CONFLICT (timer_id) DO NOTHING
"""
( contrazip6
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.timestamptz))
(E.param (E.nonNullable E.jsonb))
(E.param (E.nonNullable E.text))
)
((> 0) <$> D.rowsAffected)
claimDueTimerStmt :: Statement UTCTime (Maybe TimerRow)
claimDueTimerStmt =
preparable
"""
WITH due AS (
SELECT timer_id
FROM keiro.keiro_timers
WHERE status = 'scheduled'
AND fire_at <= $1
ORDER BY fire_at, timer_id
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE keiro.keiro_timers kt
SET status = 'firing',
attempts = kt.attempts + 1,
updated_at = now()
FROM due
WHERE kt.timer_id = due.timer_id
RETURNING kt.timer_id, kt.process_manager_name, kt.correlation_id, kt.fire_at,
kt.payload, kt.status, kt.attempts, kt.fired_event_id
"""
(E.param (E.nonNullable E.timestamptz))
(D.rowMaybe timerRowDecoder)
lookupTimerStmt :: Statement UUID (Maybe TimerRow)
lookupTimerStmt =
preparable
"""
SELECT timer_id, process_manager_name, correlation_id, fire_at,
payload, status, attempts, fired_event_id
FROM keiro.keiro_timers
WHERE timer_id = $1
"""
(E.param (E.nonNullable E.uuid))
(D.rowMaybe timerRowDecoder)
lookupTimerInspectionStmt :: Statement UUID (Maybe TimerInspection)
lookupTimerInspectionStmt =
preparable
"""
SELECT timer_id, process_manager_name, correlation_id, fire_at,
payload, status, attempts, fired_event_id, last_error
FROM keiro.keiro_timers
WHERE timer_id = $1
"""
(E.param (E.nonNullable E.uuid))
(D.rowMaybe timerInspectionDecoder)
findDeadTimersStmt :: Statement (Maybe Text, Int32, Maybe Text, Maybe UUID, Int64) [TimerInspection]
findDeadTimersStmt =
preparable
"""
SELECT timer_id, process_manager_name, correlation_id, fire_at,
payload, status, attempts, fired_event_id, last_error
FROM keiro.keiro_timers
WHERE status = 'dead'
AND ($1::text IS NULL OR process_manager_name COLLATE "C" = $1)
AND (CASE $2::integer
WHEN 0 THEN TRUE
WHEN 1 THEN last_error IS NULL
WHEN 2 THEN last_error COLLATE "C" = $3::text
WHEN 3 THEN left(last_error, char_length($3::text)) COLLATE "C" = $3::text
ELSE FALSE END)
AND ($4::uuid IS NULL OR timer_id > $4)
ORDER BY timer_id ASC
LIMIT $5::bigint
"""
( contrazip5
(E.param (E.nullable E.text))
(E.param (E.nonNullable E.int4))
(E.param (E.nullable E.text))
(E.param (E.nullable E.uuid))
(E.param (E.nonNullable E.int8))
)
(D.rowList timerInspectionDecoder)
timerInspectionDecoder :: D.Row TimerInspection
timerInspectionDecoder =
TimerInspection <$> timerRowDecoder <*> D.column (D.nullable D.text)
markTimerFiredStmt :: Statement (UUID, UUID) Bool
markTimerFiredStmt =
preparable
"""
UPDATE keiro.keiro_timers
SET status = 'fired',
fired_event_id = $2,
updated_at = now()
WHERE timer_id = $1
AND status = 'firing'
AND resume_claim_token IS NULL
"""
( contrazip2
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.uuid))
)
((> 0) <$> D.rowsAffected)
countDueTimersStmt :: Statement UTCTime Int
countDueTimersStmt =
preparable
"""
SELECT count(*)
FROM keiro.keiro_timers
WHERE status = 'scheduled'
AND fire_at <= $1
"""
(E.param (E.nonNullable E.timestamptz))
(D.singleRow (fromIntegral <$> D.column (D.nonNullable D.int8)))
countStuckTimersStmt :: Statement (Maybe UTCTime, Maybe Int64) Int
countStuckTimersStmt =
preparable
"""
SELECT count(*)
FROM keiro.keiro_timers
WHERE status = 'firing'
AND ($1::timestamptz IS NULL OR updated_at <= $1)
AND ($2::bigint IS NULL OR attempts >= $2)
"""
( contrazip2
(E.param (E.nullable E.timestamptz))
(E.param (E.nullable E.int8))
)
(D.singleRow (fromIntegral <$> D.column (D.nonNullable D.int8)))
findStuckTimersStmt :: Statement (Maybe UTCTime, Maybe Int64) [TimerRow]
findStuckTimersStmt =
preparable
"""
SELECT timer_id, process_manager_name, correlation_id, fire_at,
payload, status, attempts, fired_event_id
FROM keiro.keiro_timers
WHERE status = 'firing'
AND ($1::timestamptz IS NULL OR updated_at <= $1)
AND ($2::bigint IS NULL OR attempts >= $2)
ORDER BY updated_at, timer_id
"""
( contrazip2
(E.param (E.nullable E.timestamptz))
(E.param (E.nullable E.int8))
)
(D.rowList timerRowDecoder)
requeueStuckTimersStmt :: Statement UTCTime Int
requeueStuckTimersStmt =
preparable
"""
UPDATE keiro.keiro_timers
SET status = 'scheduled',
updated_at = now()
WHERE status = 'firing'
AND resume_claim_token IS NULL
AND updated_at <= $1
"""
(E.param (E.nonNullable E.timestamptz))
(fromIntegral <$> D.rowsAffected)
requeueStuckTimerStmt :: Statement UUID Bool
requeueStuckTimerStmt =
preparable
"""
UPDATE keiro.keiro_timers
SET status = 'scheduled',
updated_at = now()
WHERE timer_id = $1
AND status = 'firing'
AND resume_claim_token IS NULL
"""
(E.param (E.nonNullable E.uuid))
((> 0) <$> D.rowsAffected)
cancelTimerStmt :: Statement UUID Bool
cancelTimerStmt =
preparable
"""
UPDATE keiro.keiro_timers
SET status = 'cancelled',
updated_at = now()
WHERE timer_id = $1
AND status IN ('scheduled', 'firing')
AND resume_claim_token IS NULL
"""
(E.param (E.nonNullable E.uuid))
((> 0) <$> D.rowsAffected)
deadLetterTimerStmt :: Statement (UUID, Text) Bool
deadLetterTimerStmt =
preparable
"""
UPDATE keiro.keiro_timers
SET status = 'dead',
last_error = $2,
updated_at = now()
WHERE timer_id = $1
AND status IN ('scheduled', 'firing')
AND resume_claim_token IS NULL
"""
( contrazip2
(E.param (E.nonNullable E.uuid))
(E.param (E.nonNullable E.text))
)
((> 0) <$> D.rowsAffected)
timerRowDecoder :: D.Row TimerRow
timerRowDecoder =
TimerRow
<$> (TimerId <$> D.column (D.nonNullable D.uuid))
<*> D.column (D.nonNullable D.text)
<*> D.column (D.nonNullable D.text)
<*> D.column (D.nonNullable D.timestamptz)
<*> D.column (D.nonNullable D.jsonb)
<*> D.column (D.nonNullable (D.refine statusFromText D.text))
<*> (fromIntegral <$> D.column (D.nonNullable D.int8))
<*> (fmap EventId <$> D.column (D.nullable D.uuid))
statusToText :: TimerStatus -> Text
statusToText = \case
Scheduled -> "scheduled"
Firing -> "firing"
Fired -> "fired"
Cancelled -> "cancelled"
Dead -> "dead"
statusFromText :: Text -> Either Text TimerStatus
statusFromText = \case
"scheduled" -> Right Scheduled
"firing" -> Right Firing
"fired" -> Right Fired
"cancelled" -> Right Cancelled
"dead" -> Right Dead
other -> Left ("unknown keiro_timers.status: " <> other)
timerIdToUuid :: TimerId -> UUID
timerIdToUuid (TimerId uuid) = uuid
eventIdToUuid :: EventId -> UUID
eventIdToUuid (EventId uuid) = uuid