Skip to content

Fix exception handling when fetching jobs #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Raveline wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Fix exception handling when fetching jobs #43

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
47ce114
Ensure we handle only DBException
Raveline Apr 2, 2026
2f6a516
Limit the scope of the initial handle
Raveline Apr 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 12 additions & 12 deletions consumers/src/Database/PostgreSQL/Consumers/Components.hs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -377,13 +377,13 @@ spawnDispatcher ConsumerConfig {..} cs cid semaphore runningJobsInfo runningJobs
jobIds :: [idx] <- fetchMany runIdentity
if null jobIds
then pure []
else
else do
handle
( \(SomeException e) -> do
( \(DBException _ _ e _) -> do
logAttention "Failure to fetch the jobs, will reenqueue for the next day" $ object ["error" .= show e, "job_ids" .= show jobIds]
rollback
lift $ fetchFailureHandler jobIds
pure []
pure ()
)
( do
runPreparedSQL_ (preparedSqlName "setReservation" ccJobsTable) $
Expand All @@ -397,16 +397,16 @@ spawnDispatcher ConsumerConfig {..} cs cid semaphore runningJobsInfo runningJobs
, "WHERE id = ANY(" <?> Array1 jobIds <+> ")"
, "RETURNING id, " <+> mintercalate ", " ccJobSelectors
]
qr <- queryResult
results <- forM (F.toList qr) $ \(rawJobId :*: other) -> do
let jobId = runIdentity rawJobId
(Just <$> liftBase (evaluate $ ccJobFetcher other)) `catch` \(SomeException e) -> do
logAttention "Failure to fetch job, will reenqueue for the next day" $ object ["error" .= show e, "job_id" .= show jobId]
rollback
lift $ fetchFailureHandler [jobId]
pure Nothing
pure $ catMaybes results
)
qr <- queryResult
results <- forM (F.toList qr) $ \(rawJobId :*: other) -> do
let jobId = runIdentity rawJobId
(Just <$> liftBase (evaluate $ ccJobFetcher other)) `catch` \(DBException _ _ e _) -> do
logAttention "Failure to fetch job, will reenqueue for the next day" $ object ["error" .= show e, "job_id" .= show jobId]
rollback
lift $ fetchFailureHandler [jobId]
pure Nothing
pure $ catMaybes results

fetchFailureHandler :: [idx] -> m ()
fetchFailureHandler jobIds = do
Expand Down
Loading