diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,30 @@ [Keep a Changelog](https://keepachangelog.com/), and the project aims to follow the [Haskell Package Versioning Policy](https://pvp.haskell.org/). -## Unreleased +## [Unreleased] + +## 0.16.0.0 — 2026-09-07 + +### Breaking Changes + +- Ordinary worker passes recover expired foreground claims even when ordinary + requeueing is disabled. ID-only timer mutations refuse guarded claims. + Existing rows and callback signatures remain source-compatible. +- Stop/drain all old timer writers before migration 0032 and deploy upgraded + writers before enabling resume. Mixed-version writers are unsafe. Disable + resume and drain/recover claims before rollback. External effects remain + at-least-once and require consumer-owned idempotency. + +### New Features + +- Add guarded Dead timer claims with exact owner/reason checks, total attempt + ceilings, and opaque expiring ownership. Complete, renew, park, cancel, or + recover directly to Dead while retaining original work and retry history. + +- Add `Keiro.Timer.lookupTimerInspection` with full nullable stored reasons and + `findDeadTimers` with exact owner/literal reason filters, 1–100 row bounds, + and exclusive UUID pagination. Existing timer rows and worker APIs remain + compatible; callers own payload authorization before rendering. ## 0.15.0.0 — 2026-08-30 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.15.0.0 +version: 0.16.0.0 synopsis: Event sourcing framework and workflow engine description: A library that composes kiroku, keiki, and shibuya into an @@ -147,7 +147,7 @@ , hs-opentelemetry-semantic-conventions >=1.40 && <2 , keiki >=0.9 && <0.10 , keiki-codec-json >=0.9 && <0.10 - , keiro-core ^>=0.15.0.0 + , keiro-core ^>=0.16.0.0 , kiroku-store >=0.8 && <0.9 , lens >=5.2 && <5.4 , mmzk-typeid >=0.7 && <0.8 @@ -205,7 +205,7 @@ , keiki , keiki-codec-json , keiro - , keiro-test-support ^>=0.15.0.0 + , keiro-test-support ^>=0.16.0.0 , kiroku-store >=0.8 && <0.9 , process >=1.6 && <1.7 , shibuya-core ^>=0.9.0.0 @@ -236,8 +236,8 @@ , hs-opentelemetry-sdk >=1.0 && <1.1 , keiki >=0.9 && <0.10 , keiro - , keiro-core ^>=0.15.0.0 - , keiro-test-support ^>=0.15.0.0 + , keiro-core ^>=0.16.0.0 + , keiro-test-support ^>=0.16.0.0 , kiroku-store >=0.8 && <0.9 , shibuya-core ^>=0.9.0.0 , streamly-core >=0.3 && <0.4 diff --git a/src/Keiro/Timer.hs b/src/Keiro/Timer.hs --- a/src/Keiro/Timer.hs +++ b/src/Keiro/Timer.hs @@ -9,6 +9,14 @@ -- timer left @Firing@ by a crash becomes claimable again after the worker's -- configured stale-claim timeout, giving at-least-once firing. -- +-- Guarded foreground resumes use 'claimDeadTimer' after consumer authorization +-- and session preflight. Renew during work, then complete, park, or cancel with +-- the opaque handle. Expired claims recover directly to Dead, retaining reason +-- and attempts; even workers with ordinary recovery disabled perform this sweep. +-- ID-only mutations refuse guarded claims. External work remains at-least-once; +-- callers own cancellation and durable result deduplication. Upgrade every timer +-- writer before enabling resume; old binaries do not enforce the token guards. +-- -- The wire types live in "Keiro.Timer.Types" and the SQL storage in -- "Keiro.Timer.Schema"; both are re-exported here so most callers need only -- import @Keiro.Timer@. @@ -19,6 +27,30 @@ TimerRow (..), TimerStatus (..), + -- * Read-only inspection + TimerInspection (..), + TimerReasonFilter (..), + DeadTimerFilter (..), + anyDeadTimer, + DeadTimerPageRequest (..), + DeadTimerReadError (..), + DeadTimerPage (..), + lookupTimerInspection, + findDeadTimers, + + -- * Guarded foreground resume + DeadTimerClaimRequest (..), + TimerResumeError (..), + TimerResumeClaim, + resumeClaimTimer, + resumeClaimLeaseUntil, + claimDeadTimer, + renewTimerResume, + completeTimerResume, + parkTimerResume, + cancelTimerResume, + recoverExpiredTimerResumes, + -- * Storage scheduleTimerTx, scheduleTimerOnceTx, @@ -76,7 +108,7 @@ -- @updated_at@ is at least @ttl@ old back to 'Scheduled'. A fire action that -- runs longer than this timeout may be fired again; timer handlers must be -- idempotent under keiro's at-least-once timer contract. @Nothing@ disables - -- automatic requeue for callers that run their own recovery. + -- automatic ordinary requeue. Expired foreground recovery always runs. requeueStuckAfter :: !(Maybe NominalDiffTime) } deriving stock (Generic, Eq, Show) @@ -142,6 +174,7 @@ UTCTime -> Eff es () timerPassPreamble metrics options now = do + void recoverExpiredTimerResumes for_ (options ^. #requeueStuckAfter) $ \ttl -> do requeued <- requeueStuckTimers ttl now recordTimerRequeued metrics (fromIntegral requeued) diff --git a/src/Keiro/Timer/Schema.hs b/src/Keiro/Timer/Schema.hs --- a/src/Keiro/Timer/Schema.hs +++ b/src/Keiro/Timer/Schema.hs @@ -8,6 +8,9 @@ -- 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 @@ -15,6 +18,30 @@ TimerStatus (..), TimerRow (..), + -- * Read-only inspection + TimerInspection (..), + TimerReasonFilter (..), + DeadTimerFilter (..), + anyDeadTimer, + DeadTimerPageRequest (..), + DeadTimerReadError (..), + DeadTimerPage (..), + lookupTimerInspection, + findDeadTimers, + + -- * Guarded foreground resume + DeadTimerClaimRequest (..), + TimerResumeError (..), + TimerResumeClaim, + resumeClaimTimer, + resumeClaimLeaseUntil, + claimDeadTimer, + renewTimerResume, + completeTimerResume, + parkTimerResume, + cancelTimerResume, + recoverExpiredTimerResumes, + -- * Storage scheduleTimerTx, scheduleTimerOnceTx, @@ -37,10 +64,12 @@ ) where -import Contravariant.Extras (contrazip2, contrazip6) +import Contravariant.Extras (contrazip2, contrazip5, contrazip6) +import Data.Int (Int32) import Data.Time (NominalDiffTime, addUTCTime) import Data.UUID (UUID) -import Effectful (Eff, (:>)) +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) @@ -58,7 +87,7 @@ -- claimable again when 'requeueStuckTimers' moves them back to 'Scheduled'. -- * 'Fired' — successfully fired; terminal. -- * 'Cancelled' — withdrawn before firing. --- * 'Dead' — abandoned after exceeding the attempt ceiling; terminal; carries an +-- * 'Dead' — parked or abandoned; guarded foreground resume is possible; carries an -- optional @last_error@ describing why it was given up on. data TimerStatus = Scheduled @@ -85,6 +114,280 @@ } 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 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 @@ -221,7 +524,7 @@ -- | 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 --- (@SELECT * FROM keiro_timers WHERE status = 'dead'@). Terminal rows are left +-- 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 = @@ -312,6 +615,50 @@ (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 @@ -322,6 +669,7 @@ updated_at = now() WHERE timer_id = $1 AND status = 'firing' + AND resume_claim_token IS NULL """ ( contrazip2 (E.param (E.nonNullable E.uuid)) @@ -383,6 +731,7 @@ 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)) @@ -397,6 +746,7 @@ 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) @@ -410,6 +760,7 @@ 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) @@ -424,6 +775,7 @@ 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)) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -11,6 +11,7 @@ import Control.Concurrent.MVar (MVar, modifyMVar, newEmptyMVar, newMVar, putMVar, readMVar, takeMVar, tryPutMVar) import Control.Concurrent.STM (atomically, putTMVar) import Control.Exception (ErrorCall, Exception, SomeException, displayException, evaluate, finally, throwIO, try) +import Control.Monad (forM, forM_) import Data.Aeson (object, withObject, (.:), (.:?)) import Data.Aeson qualified as Aeson import Data.Aeson.KeyMap qualified as KeyMap @@ -22,6 +23,7 @@ import Data.List (isInfixOf) import Data.List.NonEmpty qualified as NonEmpty import Data.Map.Strict qualified as Map +import Data.Maybe (catMaybes) import Data.Monoid (mempty) import Data.Set qualified as Set import Data.Text qualified as Text @@ -215,6 +217,7 @@ import Keiro.Telemetry qualified as Telemetry import Keiro.Test.Postgres ( StoreRunner (..), + withFreshDatabase, withFreshResourceStore, withFreshResourceStoreWith, withFreshStore, @@ -223,6 +226,7 @@ withMigratedSuite, ) import Keiro.Timer +import Keiro.Timer qualified as Timer import Keiro.Wake ( WakeReason (..), WakeSignal (..), @@ -5507,7 +5511,388 @@ (results, Vector.length targetEvents) `shouldBe` ([DomainPMCommandDuplicate legacyId], 1) + describe "Keiro.Timer foreground consumer" + $ around + ( \action -> + withFreshDatabase fixture $ \connection -> + Store.withStore (Store.defaultConnectionSettings connection) $ \firstStore -> + Store.withStore (Store.defaultConnectionSettings connection) $ \secondStore -> + action (firstStore, secondStore) + ) + $ do + it "preflights original work and invokes one callback for competing authorized resumes" $ \(firstStore, secondStore) -> do + let original = counterTimerRequest & #payload .~ object ["memorySpace" Aeson..= ("space-a" :: Text)] + tid = original ^. #timerId + owner = original ^. #processManagerName + reason = "deferred: interactive session required" + callbacks <- newIORef (0 :: Int) + let foreground store allowed available = do + Right _ <- Store.runStoreIO store recoverExpiredTimerResumes + Right inspected <- Store.runStoreIO store $ lookupTimerInspection tid + case inspected of + Just inspection + | inspection ^. #lastError == Just reason, + inspection ^. #timer . #payload == object ["memorySpace" Aeson..= ("space-a" :: Text)], + allowed, + available -> do + Right (Right claimed) <- Store.runStoreIO store $ claimDeadTimer (DeadTimerClaimRequest tid owner reason 3 60) + forM_ claimed $ \_ -> atomicModifyIORef' callbacks (\n -> (n + 1, ())) + pure claimed + _ -> pure Nothing + Right () <- Store.runStoreIO firstStore $ Store.runTransaction $ scheduleTimerTx original + Right True <- Store.runStoreIO firstStore $ deadLetterTimer tid reason + Right before <- Store.runStoreIO firstStore $ Store.runTransaction $ Tx.statement () timerReadSnapshotStmt + -- Revocation after listing and repeated unavailable-session preflights. + Right (Right _) <- Store.runStoreIO firstStore $ findDeadTimers (DeadTimerFilter (Just owner) (ReasonExact reason)) (DeadTimerPageRequest 10 Nothing) + denied <- foreground firstStore False True + isNothing denied `shouldBe` True + forM_ [1 .. 3 :: Int] $ \_ -> do + unavailable <- foreground firstStore True False + isNothing unavailable `shouldBe` True + Store.runStoreIO firstStore (Store.runTransaction (Tx.statement () timerReadSnapshotStmt)) `shouldReturn` Right before + (a, b) <- timerRaceIO (foreground firstStore True True) (foreground secondStore True True) + length (catMaybes [a, b]) `shouldBe` 1 + readIORef callbacks `shouldReturn` 1 + -- Crash, deterministic expiry, and recovery keep interactive work parked. + Right () <- Store.runStoreIO firstStore $ Store.runTransaction expireTimerResumesTx + Store.runStoreIO secondStore recoverExpiredTimerResumes `shouldReturn` Right 1 + Store.runStoreIO firstStore (runTimerWorker Nothing dueTimerTime (\_ -> error "interactive work dispatched in background")) `shouldReturn` Right Nothing + unavailable <- foreground firstStore True False + isNothing unavailable `shouldBe` True + Just next <- foreground secondStore True True + resumeClaimTimer next ^. #timerId `shouldBe` tid + resumeClaimTimer next ^. #attempts `shouldBe` 2 + -- A transient post-claim failure consumes the attempt and retains reason. + Store.runStoreIO secondStore (parkTimerResume next) `shouldReturn` Right True + Right (Just parked) <- Store.runStoreIO firstStore $ lookupTimerInspection tid + parked ^. #lastError `shouldBe` Just reason + parked ^. #timer . #attempts `shouldBe` 2 + -- Malformed work and ordinary dead letters are application refusals. + Right () <- Store.runStoreIO firstStore $ Store.runTransaction $ Tx.sql "UPDATE keiro.keiro_timers SET payload = '{}'::jsonb" + malformed <- foreground firstStore True True + isNothing malformed `shouldBe` True + Right () <- Store.runStoreIO firstStore $ Store.runTransaction $ Tx.sql "UPDATE keiro.keiro_timers SET last_error = 'ordinary dead letter'" + ordinary <- foreground firstStore True True + isNothing ordinary `shouldBe` True + readIORef callbacks `shouldReturn` 2 + + it "orders renewal and completion against recovery on independent stores" $ \(firstStore, secondStore) -> do + let tid = counterTimerRequest ^. #timerId + request = DeadTimerClaimRequest tid (counterTimerRequest ^. #processManagerName) "deferred" 3 60 + Right () <- Store.runStoreIO firstStore $ Store.runTransaction $ scheduleTimerTx counterTimerRequest + Right True <- Store.runStoreIO firstStore $ deadLetterTimer tid "deferred" + Right (Right (Just claim)) <- Store.runStoreIO firstStore $ claimDeadTimer request + (renewed, recovered) <- + timerRaceIO + (Store.runStoreIO firstStore $ renewTimerResume claim 60) + (Store.runStoreIO secondStore recoverExpiredTimerResumes) + renewed `shouldBe` Right (Right True) + recovered `shouldBe` Right 0 + Right () <- Store.runStoreIO firstStore $ Store.runTransaction expireTimerResumesTx + (expiredRenewal, expiredRecovery) <- + timerRaceIO + (Store.runStoreIO firstStore $ renewTimerResume claim 60) + (Store.runStoreIO secondStore recoverExpiredTimerResumes) + expiredRenewal `shouldBe` Right (Right False) + expiredRecovery `shouldBe` Right 1 + Right (Right (Just replacement)) <- Store.runStoreIO firstStore $ claimDeadTimer request + Right before <- Store.runStoreIO firstStore $ Store.runTransaction $ Tx.statement () timerReadSnapshotStmt + forM_ [parkTimerResume claim, cancelTimerResume claim, completeTimerResume claim (EventId sampleUuid2)] $ \operation -> + Store.runStoreIO firstStore operation `shouldReturn` Right False + Store.runStoreIO firstStore (Store.runTransaction (Tx.statement () timerReadSnapshotStmt)) `shouldReturn` Right before + Right () <- Store.runStoreIO firstStore $ Store.runTransaction expireTimerResumesTx + (expiredCompletion, completionRecovery) <- + timerRaceIO + (Store.runStoreIO firstStore $ completeTimerResume replacement (EventId sampleUuid2)) + (Store.runStoreIO secondStore recoverExpiredTimerResumes) + expiredCompletion `shouldBe` Right False + completionRecovery `shouldBe` Right 1 + Right (Right (Just finalClaim)) <- Store.runStoreIO firstStore $ claimDeadTimer request + (completed, noRecovery) <- + timerRaceIO + (Store.runStoreIO firstStore $ completeTimerResume finalClaim (EventId sampleUuid2)) + (Store.runStoreIO secondStore recoverExpiredTimerResumes) + completed `shouldBe` Right True + noRecovery `shouldBe` Right 0 + describe "Keiro.Timer" $ around (withFreshStore fixture) $ do + it "guards dead resume ownership and retains attempts when parked" $ \storeHandle -> do + let tid = counterTimerRequest ^. #timerId + request = DeadTimerClaimRequest tid (counterTimerRequest ^. #processManagerName) "deferred" 2 60 + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx counterTimerRequest + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer tid "deferred" + Right (Right (Just claim)) <- Store.runStoreIO storeHandle $ claimDeadTimer request + resumeClaimTimer claim ^. #attempts `shouldBe` 1 + Store.runStoreIO storeHandle (claimDueTimer dueTimerTime) `shouldReturn` Right Nothing + Store.runStoreIO storeHandle (markTimerFired tid (EventId sampleUuid2)) `shouldReturn` Right False + Store.runStoreIO storeHandle (cancelTimer tid) `shouldReturn` Right False + Store.runStoreIO storeHandle (deadLetterTimer tid "wrong") `shouldReturn` Right False + Store.runStoreIO storeHandle (requeueStuckTimer tid) `shouldReturn` Right False + Store.runStoreIO storeHandle (requeueStuckTimers 0 (resumeClaimLeaseUntil claim)) `shouldReturn` Right 0 + Store.runStoreIO storeHandle (renewTimerResume claim 60) `shouldReturn` Right (Right True) + Store.runStoreIO storeHandle (parkTimerResume claim) `shouldReturn` Right True + Right (Right (Just replacement)) <- Store.runStoreIO storeHandle $ claimDeadTimer request + resumeClaimTimer replacement ^. #attempts `shouldBe` 2 + Store.runStoreIO storeHandle (completeTimerResume claim (EventId sampleUuid2)) `shouldReturn` Right False + Store.runStoreIO storeHandle (parkTimerResume claim) `shouldReturn` Right False + Store.runStoreIO storeHandle (cancelTimerResume claim) `shouldReturn` Right False + Store.runStoreIO storeHandle (renewTimerResume claim 60) `shouldReturn` Right (Right False) + Store.runStoreIO storeHandle (parkTimerResume replacement) `shouldReturn` Right True + Right (Right refused) <- Store.runStoreIO storeHandle $ claimDeadTimer request + isNothing refused `shouldBe` True + Right (Just observed) <- Store.runStoreIO storeHandle $ lookupTimerInspection tid + observed ^. #lastError `shouldBe` Just "deferred" + observed ^. #timer . #attempts `shouldBe` 2 + + it "refuses every ineligible claim without changing any persisted column" $ \storeHandle -> do + let tid = counterTimerRequest ^. #timerId + request = DeadTimerClaimRequest tid (counterTimerRequest ^. #processManagerName) "deferred" 1 60 + snapshot = Store.runStoreIO storeHandle $ Store.runTransaction $ Tx.statement () timerReadSnapshotStmt + refused req = do + before <- snapshot + Right (Right result) <- Store.runStoreIO storeHandle $ claimDeadTimer req + isNothing result `shouldBe` True + snapshot `shouldReturn` before + refused request + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx counterTimerRequest + refused request + Right (Just _) <- Store.runStoreIO storeHandle $ claimDueTimer dueTimerTime + refused request + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer tid "deferred" + refused (request & #processManagerName .~ "COUNTER") + refused (request & #expectedReason .~ "Deferred") + refused (request & #maxAttempts .~ 0) + refused request -- ordinary claim already consumed the ceiling + before <- snapshot + Right (Left badMax) <- Store.runStoreIO storeHandle $ claimDeadTimer (request & #maxAttempts .~ (-1)) + badMax `shouldBe` InvalidTimerResumeMaxAttempts (-1) + forM_ [0, -1, maxBound] $ \seconds -> do + Right (Left badLease) <- Store.runStoreIO storeHandle $ claimDeadTimer (request & #leaseSeconds .~ seconds) + badLease `shouldBe` InvalidTimerResumeLeaseSeconds seconds + snapshot `shouldReturn` before + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ Tx.statement (case tid of TimerId uuid -> uuid) legacyDeadTimerReasonStmt + refused (request & #maxAttempts .~ 2) + + it "claims literal empty and Unicode reasons and preserves original work" $ \storeHandle -> do + forM_ (zip [1 ..] ["", "延期: café 日本語 🌱", "a%_\\'雪"]) $ \(n, reason) -> do + let original = counterTimerRequest & #timerId .~ TimerId (UUID.fromWords 0 0 0 n) + tid = original ^. #timerId + request = DeadTimerClaimRequest tid (original ^. #processManagerName) reason 1 60 + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx original + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer tid reason + Right (Just before) <- Store.runStoreIO storeHandle $ lookupTimer tid + Right (Right (Just claim)) <- Store.runStoreIO storeHandle $ claimDeadTimer request + resumeClaimTimer claim `shouldBe` (before & #status .~ Firing & #attempts .~ 1) + Right (Right repeated) <- Store.runStoreIO storeHandle $ claimDeadTimer request + isNothing repeated `shouldBe` True + Store.runStoreIO storeHandle (completeTimerResume claim (EventId sampleUuid2)) `shouldReturn` Right True + Right (Just inspection) <- Store.runStoreIO storeHandle $ lookupTimerInspection tid + inspection ^. #lastError `shouldBe` Just reason + inspection ^. #timer . #firedEventId `shouldBe` Just (EventId sampleUuid2) + Right (Right terminal) <- Store.runStoreIO storeHandle $ claimDeadTimer (request & #maxAttempts .~ 2) + isNothing terminal `shouldBe` True + + it "expires without revival and re-parks independently of ordinary recovery" $ \storeHandle -> do + let tid = counterTimerRequest ^. #timerId + request = DeadTimerClaimRequest tid (counterTimerRequest ^. #processManagerName) "deferred" 3 60 + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx counterTimerRequest + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer tid "deferred" + Right (Right (Just old)) <- Store.runStoreIO storeHandle $ claimDeadTimer request + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction expireTimerResumesTx + Right before <- Store.runStoreIO storeHandle $ Store.runTransaction $ Tx.statement () timerReadSnapshotStmt + Store.runStoreIO storeHandle (renewTimerResume old 60) `shouldReturn` Right (Right False) + Store.runStoreIO storeHandle (completeTimerResume old (EventId sampleUuid2)) `shouldReturn` Right False + Store.runStoreIO storeHandle (parkTimerResume old) `shouldReturn` Right False + Store.runStoreIO storeHandle (cancelTimerResume old) `shouldReturn` Right False + Store.runStoreIO storeHandle (Store.runTransaction (Tx.statement () timerReadSnapshotStmt)) `shouldReturn` Right before + let options = defaultTimerWorkerOptions & #requeueStuckAfter .~ Nothing + Store.runStoreIO storeHandle (runTimerWorkerWith Nothing options dueTimerTime (\_ -> error "foreground work reached background")) `shouldReturn` Right Nothing + Store.runStoreIO storeHandle recoverExpiredTimerResumes `shouldReturn` Right 0 + Right (Just parked) <- Store.runStoreIO storeHandle $ lookupTimerInspection tid + parked ^. #timer . #status `shouldBe` Dead + parked ^. #timer . #attempts `shouldBe` 1 + parked ^. #lastError `shouldBe` Just "deferred" + Right (Right (Just replacement)) <- Store.runStoreIO storeHandle $ claimDeadTimer request + Store.runStoreIO storeHandle (completeTimerResume old (EventId sampleUuid2)) `shouldReturn` Right False + Store.runStoreIO storeHandle (renewTimerResume old 60) `shouldReturn` Right (Right False) + Store.runStoreIO storeHandle (cancelTimerResume replacement) `shouldReturn` Right True + Right (Right cancelled) <- Store.runStoreIO storeHandle $ claimDeadTimer request + isNothing cancelled `shouldBe` True + + it "inspects absent timers and preserves legacy metadata through every lifecycle" $ \storeHandle -> do + let tid = counterTimerRequest ^. #timerId + inspect reason = do + Right old <- Store.runStoreIO storeHandle $ lookupTimer tid + Right observed <- Store.runStoreIO storeHandle $ lookupTimerInspection tid + fmap (^. #timer) observed `shouldBe` old + fmap (^. #lastError) observed `shouldBe` Just reason + Store.runStoreIO storeHandle (lookupTimerInspection tid) `shouldReturn` Right Nothing + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx counterTimerRequest + inspect Nothing + Right (Just _) <- Store.runStoreIO storeHandle $ claimDueTimer dueTimerTime + inspect Nothing + Right True <- Store.runStoreIO storeHandle $ markTimerFired tid (EventId sampleUuid2) + inspect Nothing + Right (Just observed) <- Store.runStoreIO storeHandle $ lookupTimerInspection tid + observed ^. #timer . #firedEventId `shouldBe` Just (EventId sampleUuid2) + observed ^. #timer . #attempts `shouldBe` 1 + + it "preserves empty, populated, and Unicode dead reasons verbatim" $ \storeHandle -> do + forM_ (zip [1 ..] ["", " retry exhausted ", "延期: café 日本語 🌱"]) $ \(n, reason) -> do + let request = counterTimerRequest & #timerId .~ TimerId (UUID.fromWords 0 0 0 n) + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx request + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer (request ^. #timerId) reason + Right old <- Store.runStoreIO storeHandle $ lookupTimer (request ^. #timerId) + Right (Just observed) <- Store.runStoreIO storeHandle $ lookupTimerInspection (request ^. #timerId) + Just (observed ^. #timer) `shouldBe` old + observed ^. #lastError `shouldBe` Just reason + + it "filters dead timers by exact owner and literal reason, preserving NULL" $ \storeHandle -> do + let fixtures = + [ (1, "A", Just "deferred: one"), + (2, "A", Just "deferred: 二"), + (3, "B", Just "deferred: three"), + (4, "A", Just "ordinary"), + (5, "A", Nothing), + (6, "A", Just ""), + (7, "A", Just "a%_\\'雪 tail"), + (8, "A", Just "aXX雪 tail"), + (9, "a", Just "Deferred: one") + ] + tid n = TimerId (UUID.fromWords 0 0 0 n) + check owner reason expected = do + Right (Right page) <- + Store.runStoreIO storeHandle $ + findDeadTimers (DeadTimerFilter owner reason) (DeadTimerPageRequest 100 Nothing) + fmap (^. #timer . #timerId) (page ^. #timers) `shouldBe` fmap tid expected + page ^. #nextAfterTimerId `shouldBe` Nothing + forM_ fixtures $ \(n, owner, reason) -> do + let request = counterTimerRequest & #timerId .~ tid n & #processManagerName .~ owner + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx request + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer (tid n) (fromMaybe "legacy" reason) + when (isNothing reason) $ do + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + Tx.statement (UUID.fromWords 0 0 0 n) legacyDeadTimerReasonStmt + Right (Just inspection) <- Store.runStoreIO storeHandle $ lookupTimerInspection (tid n) + inspection ^. #lastError `shouldBe` Nothing + -- Include every non-dead lifecycle in the same manager/reason search space. + forM_ [10 .. 13] $ \n -> do + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + scheduleTimerTx (counterTimerRequest & #timerId .~ tid n & #processManagerName .~ "A") + pure () + Right (Just _) <- Store.runStoreIO storeHandle $ claimDueTimer dueTimerTime + Right True <- Store.runStoreIO storeHandle $ markTimerFired (tid 10) (EventId sampleUuid2) + Right (Just _) <- Store.runStoreIO storeHandle $ claimDueTimer dueTimerTime + Right True <- Store.runStoreIO storeHandle $ cancelTimer (tid 12) + Right (Just cancelled) <- Store.runStoreIO storeHandle $ lookupTimerInspection (tid 12) + cancelled ^. #timer . #status `shouldBe` Timer.Cancelled + check Nothing AnyTimerReason [1 .. 9] + check (Just "A") (ReasonPrefix "deferred:") [1, 2] + check Nothing (ReasonPrefix "deferred:") [1, 2, 3] + check (Just "a") AnyTimerReason [9] + check Nothing ReasonAbsent [5] + check Nothing (ReasonExact "") [6] + check Nothing (ReasonPrefix "") [1, 2, 3, 4, 6, 7, 8, 9] + check Nothing (ReasonExact "deferred: 二") [2] + check Nothing (ReasonPrefix "a%_\\'雪") [7] + check Nothing (ReasonExact "a%_\\'雪 tail") [7] + check Nothing (ReasonExact "DEFERRED: one") [] + check (Just "A' OR TRUE --") AnyTimerReason [] + + it "bounds pages, traverses UUID order, and leaves every stored column unchanged" $ \storeHandle -> do + forM_ [1 .. 101] $ \n -> do + let request = plainTimerRequest n + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx request + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer (request ^. #timerId) "deferred" + pure () + let snapshot = Store.runStoreIO storeHandle $ Store.runTransaction $ Tx.statement () timerReadSnapshotStmt + readPage size cursor = Store.runStoreIO storeHandle $ findDeadTimers anyDeadTimer (DeadTimerPageRequest size cursor) + storedBefore <- snapshot + forM_ [-1, 0, 101, maxBound] $ \size -> + readPage size Nothing `shouldReturn` Right (Left (InvalidDeadTimerPageSize size)) + Right (Right first) <- readPage 100 Nothing + length (first ^. #timers) `shouldBe` 100 + first ^. #nextAfterTimerId `shouldBe` Just (plainTimerRequest 100 ^. #timerId) + readPage 100 Nothing `shouldReturn` Right (Right first) + Right (Right finalPage) <- readPage 100 (first ^. #nextAfterTimerId) + fmap (^. #timer . #timerId) (finalPage ^. #timers) `shouldBe` [plainTimerRequest 101 ^. #timerId] + finalPage ^. #nextAfterTimerId `shouldBe` Nothing + readPage 1 (Just (plainTimerRequest 101 ^. #timerId)) `shouldReturn` Right (Right (DeadTimerPage [] Nothing)) + let walk cursor = do + Right (Right page) <- readPage 1 cursor + let ids = fmap (^. #timer . #timerId) (page ^. #timers) + case page ^. #nextAfterTimerId of + Nothing -> pure ids + next -> (ids <>) <$> walk next + walk Nothing `shouldReturn` fmap ((^. #timerId) . plainTimerRequest) [1 .. 101] + Right (Just _) <- Store.runStoreIO storeHandle $ lookupTimerInspection (plainTimerRequest 1 ^. #timerId) + snapshot `shouldReturn` storedBefore + Store.runStoreIO storeHandle (runTimerWorker Nothing dueTimerTime (\_ -> pure (Just (EventId sampleUuid2)))) + `shouldReturn` Right Nothing + snapshot `shouldReturn` storedBefore + + it "continues after a deleted cursor and observes new eligibility only above it" $ \storeHandle -> do + let add n = do + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx (plainTimerRequest n) + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer (plainTimerRequest n ^. #timerId) "deferred" + pure () + readPage cursor = Store.runStoreIO storeHandle $ findDeadTimers anyDeadTimer (DeadTimerPageRequest 1 cursor) + mapM_ add [20, 40, 60] + Right (Right first) <- readPage Nothing + first ^. #nextAfterTimerId `shouldBe` Just (plainTimerRequest 20 ^. #timerId) + Right () <- + Store.runStoreIO storeHandle $ + Store.runTransaction $ + Tx.sql + "DELETE FROM keiro.keiro_timers WHERE correlation_id = 'drain-20'; UPDATE keiro.keiro_timers SET status = 'cancelled' WHERE correlation_id = 'drain-40'" + mapM_ add [10, 30] + Right (Right second) <- readPage (first ^. #nextAfterTimerId) + fmap (^. #timer . #timerId) (second ^. #timers) `shouldBe` [plainTimerRequest 30 ^. #timerId] + Right (Right third) <- readPage (second ^. #nextAfterTimerId) + fmap (^. #timer . #timerId) (third ^. #timers) `shouldBe` [plainTimerRequest 60 ^. #timerId] + third ^. #nextAfterTimerId `shouldBe` Nothing + + it "renders authorized original work beyond empty pages and rechecks revoked permissions" $ \storeHandle -> do + let reason = "kioku:deferred:interactive-unavailable feature=summary details=保持" + deadFilter = DeadTimerFilter (Just "drain-pm") (ReasonPrefix "kioku:deferred:interactive-unavailable feature=") + entries = + [ (1, object ["space" Aeson..= ("hidden" :: Text), "work" Aeson..= ("secret" :: Text)]), + (2, object ["invalid" Aeson..= ("never render" :: Text)]), + (3, object ["space" Aeson..= ("allowed" :: Text), "work" Aeson..= ("original work" :: Text)]), + (4, object ["space" Aeson..= ("allowed" :: Text), "work" Aeson..= ("revoked work" :: Text)]) + ] + forM_ entries $ \(n, payload) -> do + Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ scheduleTimerTx (plainTimerRequest n & #payload .~ payload) + Right True <- Store.runStoreIO storeHandle $ deadLetterTimer (plainTimerRequest n ^. #timerId) reason + pure () + permissions <- newIORef (Set.singleton ("allowed" :: Text)) + let renderPage cursor = do + Right (Right page) <- Store.runStoreIO storeHandle $ findDeadTimers deadFilter (DeadTimerPageRequest 1 cursor) + fresh <- readIORef permissions + rendered <- fmap catMaybes $ forM (page ^. #timers) $ \listed -> do + Right inspected <- Store.runStoreIO storeHandle $ lookupTimerInspection (listed ^. #timer . #timerId) + pure $ do + inspection <- inspected + (space, work) <- + either (const Nothing) Just $ + parseEither (withObject "work" (\o -> (,) <$> o .: "space" <*> o .: "work")) (inspection ^. #timer . #payload) + if Set.member space fresh then Just (work :: Text, inspection ^. #lastError) else Nothing + pure (rendered, page ^. #nextAfterTimerId) + (hidden, next1) <- renderPage Nothing + hidden `shouldBe` [] + next1 `shouldSatisfy` isJust + (malformed, next2) <- renderPage next1 + malformed `shouldBe` [] + next2 `shouldSatisfy` isJust + (allowed, next3) <- renderPage next2 + allowed `shouldBe` [("original work", Just reason)] + writeIORef permissions Set.empty + renderPage next3 `shouldReturn` ([], Nothing) + it "validates worker options before startup" $ \_storeHandle -> do shouldBeRight_ (mkTimerWorkerOptions defaultTimerWorkerOptions) mkTimerWorkerOptions (defaultTimerWorkerOptions & #maxAttempts ?~ (-1)) @@ -16462,3 +16847,38 @@ (E.param (E.nonNullable E.text)) ) (D.singleRow (D.column (D.nonNullable D.int8))) + +-- Test-only legacy fixture and complete persistence snapshot, including timestamps. +legacyDeadTimerReasonStmt :: Statement UUID () +legacyDeadTimerReasonStmt = + preparable + "UPDATE keiro.keiro_timers SET last_error = NULL WHERE timer_id = $1" + (E.param (E.nonNullable E.uuid)) + D.noResult + +timerReadSnapshotStmt :: Statement () [Value] +timerReadSnapshotStmt = + preparable + "SELECT to_jsonb(t) FROM keiro.keiro_timers t ORDER BY timer_id" + E.noParams + (D.rowList (D.column (D.nonNullable D.jsonb))) + +-- Test-only deterministic expiry; foreground consumer code uses public APIs. +expireTimerResumesTx :: Tx.Transaction () +expireTimerResumesTx = Tx.sql "UPDATE keiro.keiro_timers SET resume_lease_until = clock_timestamp() - interval '1 second' WHERE resume_claim_token IS NOT NULL" + +-- Both independent connections begin only after the common barrier opens. +-- Exceptions are transported to the test thread instead of stranding its wait. +timerRaceIO :: IO a -> IO b -> IO (a, b) +timerRaceIO first second = do + start <- newEmptyMVar + a <- newEmptyMVar + b <- newEmptyMVar + let capture :: IO x -> IO (Either SomeException x) + capture = try + _ <- forkIO $ capture (readMVar start >> first) >>= putMVar a + _ <- forkIO $ capture (readMVar start >> second) >>= putMVar b + putMVar start () + ar <- takeMVar a >>= either throwIO pure + br <- takeMVar b >>= either throwIO pure + pure (ar, br)