Safe Haskell	None
Language	GHC2024

Arbiter.Core

Contents

Core types
High-level operations
Job tree DSL
Job results
Archived jobs
Typed codecs and parameters
Schema
Exceptions
Cron schedule overrides and worker health
Connection pool settings
Listener types

Description

Re-exports commonly used Arbiter functionality.

Synopsis

module Arbiter.Core.Job.DLQ
module Arbiter.Core.Job.Types
module Arbiter.Core.MonadArbiter
module Arbiter.Core.QueueRegistry
module Arbiter.Core.HighLevel
(<~~) :: JobWrite payload -> NonEmpty (JobWrite payload) -> JobTree payload
leaf :: JobWrite payload -> JobTree payload
rollup :: JobWrite payload -> NonEmpty (JobTree payload) -> JobTree payload
data JobTree payload
module Arbiter.Core.JobResult
data ArchiveJob payload = ArchiveJob {
- archivePrimaryKey :: Int64
- completedAt :: UTCTime
- jobSnapshot :: JobSnapshot payload
- archivedResult :: Maybe Value
}
module Arbiter.Core.Codec
module Arbiter.Core.Job.Schema
module Arbiter.Core.Job.Schema.Groups
module Arbiter.Core.Exceptions
data CronScheduleRow = CronScheduleRow {
- name :: Text
- queueName :: Text
- defaultExpression :: Text
- defaultOverlap :: Text
- defaultTimezone :: Maybe Text
- overrideExpression :: Maybe Text
- overrideOverlap :: Maybe Text
- overrideTimezone :: Maybe Text
- enabled :: Bool
- lastFiredAt :: Maybe UTCTime
- lastCheckedAt :: Maybe UTCTime
- runRequestedAt :: Maybe UTCTime
- lastManualRunAt :: Maybe UTCTime
- createdAt :: UTCTime
- updatedAt :: UTCTime
}
data CronScheduleUpdate = CronScheduleUpdate {
- overrideExpression :: Maybe (Maybe Text)
- overrideOverlap :: Maybe (Maybe Text)
- overrideTimezone :: Maybe (Maybe Text)
- enabled :: Maybe Bool
}
effectiveExpression :: CronScheduleRow -> Text
effectiveOverlap :: CronScheduleRow -> Text
effectiveTimezone :: CronScheduleRow -> Maybe Text
data WorkerHealth
- = Live
- | Stale
- | Draining
workerHealthFromText :: Text -> Either Text WorkerHealth
module Arbiter.Core.PoolConfig
data Listener
data RunningHub
data HubLog = HubLog {
- hubRecovered :: Text -> IO ()
- hubWarn :: Text -> IO ()
- hubError :: Text -> IO ()
- hubRepeatInterval :: NominalDiffTime
}
withChannels :: MonadUnliftIO m => Listener -> HubLog -> [(ByteString, Notification -> m ())] -> (STM Bool -> m a) -> m a
newPoolListener :: ((Connection -> IO ()) -> IO ()) -> IO Listener
data DedicatedListen
newDedicatedListen :: MonadIO m => ByteString -> m DedicatedListen
dedicatedListener :: DedicatedListen -> Listener

Core types

module Arbiter.Core.Job.DLQ

module Arbiter.Core.Job.Types

module Arbiter.Core.MonadArbiter

module Arbiter.Core.QueueRegistry

High-level operations

module Arbiter.Core.HighLevel

Job tree DSL

Source #(<~~) :: JobWrite payload -> NonEmpty (JobWrite payload) -> JobTree payload infixr 6

Infix rollup for leaf-only children.

defaultJob reducer <~~ (defaultJob mapper1 :| [defaultJob mapper2])

Source #leaf :: JobWrite payload -> JobTree payload

A single job with no children.

Source #rollup :: JobWrite payload -> NonEmpty (JobTree payload) -> JobTree payload

A finalizer running once no child of it is left in the main queue. Nested rollups do not merge on their own. An intermediate finalizer returns the merged value for results to travel upward.

rollup (defaultJob root)
  ( leaf (defaultJob leaf1)
  :| [leaf (defaultJob leaf2)]
  )

Source #data JobTree payload

A tree of jobs. Leaves are single jobs. Finalizers are parents with children that run immediately while the parent waits for completion.

Job results

module Arbiter.Core.JobResult

Archived jobs

Source #data ArchiveJob payload

A completed job in the archive.

Constructors

ArchiveJob
Fields archivePrimaryKey :: Int64 Archive table primary key. The snapshot keeps its own job id. completedAt :: UTCTime When the job was acked and archived jobSnapshot :: JobSnapshot payload Full job state at time of completion (payload, attempts, etc.) archivedResult :: Maybe Value Handler result stored for a completed root job (one with no parent).

Instances

Instances details

Eq payload => Eq (ArchiveJob payload) Source #

Instance details

Defined in Arbiter.Core.Job.Archive

Methods

#(==) :: ArchiveJob payload -> ArchiveJob payload -> Bool

#(/=) :: ArchiveJob payload -> ArchiveJob payload -> Bool

Generic (ArchiveJob payload) Source #

Instance details

Defined in Arbiter.Core.Job.Archive

Associated Types

type Rep (ArchiveJob payload)

Instance details

Defined in Arbiter.Core.Job.Archive

type Rep (ArchiveJob payload) = D1 ('MetaData "ArchiveJob" "Arbiter.Core.Job.Archive" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ArchiveJob" 'PrefixI 'True) ((S1 ('MetaSel ('Just "archivePrimaryKey") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64) :*: S1 ('MetaSel ('Just "completedAt") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 UTCTime)) :*: (S1 ('MetaSel ('Just "jobSnapshot") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (JobSnapshot payload)) :*: S1 ('MetaSel ('Just "archivedResult") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Value)))))

Methods

#from :: ArchiveJob payload -> Rep (ArchiveJob payload) x

#to :: Rep (ArchiveJob payload) x -> ArchiveJob payload

Show payload => Show (ArchiveJob payload) Source #

Instance details

Defined in Arbiter.Core.Job.Archive

Methods

#showsPrec :: Int -> ArchiveJob payload -> ShowS

#show :: ArchiveJob payload -> String

#showList :: [ArchiveJob payload] -> ShowS

type Rep (ArchiveJob payload) Source #

Instance details

Defined in Arbiter.Core.Job.Archive

type Rep (ArchiveJob payload) = D1 ('MetaData "ArchiveJob" "Arbiter.Core.Job.Archive" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ArchiveJob" 'PrefixI 'True) ((S1 ('MetaSel ('Just "archivePrimaryKey") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64) :*: S1 ('MetaSel ('Just "completedAt") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 UTCTime)) :*: (S1 ('MetaSel ('Just "jobSnapshot") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (JobSnapshot payload)) :*: S1 ('MetaSel ('Just "archivedResult") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Value)))))

Typed codecs and parameters

module Arbiter.Core.Codec

Schema

module Arbiter.Core.Job.Schema

module Arbiter.Core.Job.Schema.Groups

Exceptions

module Arbiter.Core.Exceptions

Cron schedule overrides and worker health

The admin row records share field names with each other and with Job. Unqualified use needs DuplicateRecordFields or OverloadedRecordDot.

Source #data CronScheduleRow

A row from the cron_schedules table.

Constructors

CronScheduleRow

Fields

name :: Text
queueName :: Text
defaultExpression :: Text
defaultOverlap :: Text
defaultTimezone :: Maybe Text
Code-defined IANA tz name. NULL = UTC.
overrideExpression :: Maybe Text
overrideOverlap :: Maybe Text
overrideTimezone :: Maybe Text
User override. NULL = use default. To force UTC when the default is not UTC, set to "UTC".
enabled :: Bool
lastFiredAt :: Maybe UTCTime
lastCheckedAt :: Maybe UTCTime
runRequestedAt :: Maybe UTCTime
Manual run awaiting a worker pool. NULL = none pending.
lastManualRunAt :: Maybe UTCTime
When a manual run last fired a job. NULL = never.
createdAt :: UTCTime
updatedAt :: UTCTime

Instances

Instances details

FromJSON CronScheduleRow Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#parseJSON :: Value -> Parser CronScheduleRow

#parseJSONList :: Value -> Parser [CronScheduleRow]

#omittedField :: Maybe CronScheduleRow

ToJSON CronScheduleRow Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#toJSON :: CronScheduleRow -> Value

#toEncoding :: CronScheduleRow -> Encoding

#toJSONList :: [CronScheduleRow] -> Value

#toEncodingList :: [CronScheduleRow] -> Encoding

#omitField :: CronScheduleRow -> Bool

Eq CronScheduleRow Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#(==) :: CronScheduleRow -> CronScheduleRow -> Bool

#(/=) :: CronScheduleRow -> CronScheduleRow -> Bool

Generic CronScheduleRow Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Associated Types

type Rep CronScheduleRow

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#from :: CronScheduleRow -> Rep CronScheduleRow x

#to :: Rep CronScheduleRow x -> CronScheduleRow

Show CronScheduleRow Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#showsPrec :: Int -> CronScheduleRow -> ShowS

#show :: CronScheduleRow -> String

#showList :: [CronScheduleRow] -> ShowS

type Rep CronScheduleRow Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Source #data CronScheduleUpdate

A patch over a cron schedule's overrides. Nothing leaves a field unchanged, Just Nothing clears the override back to the default, Just (Just v) sets it.

Constructors

CronScheduleUpdate
Fields overrideExpression :: Maybe (Maybe Text) overrideOverlap :: Maybe (Maybe Text) overrideTimezone :: Maybe (Maybe Text) enabled :: Maybe Bool

Instances

Instances details

FromJSON CronScheduleUpdate Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#parseJSON :: Value -> Parser CronScheduleUpdate

#parseJSONList :: Value -> Parser [CronScheduleUpdate]

#omittedField :: Maybe CronScheduleUpdate

ToJSON CronScheduleUpdate Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#toJSON :: CronScheduleUpdate -> Value

#toEncoding :: CronScheduleUpdate -> Encoding

#toJSONList :: [CronScheduleUpdate] -> Value

#toEncodingList :: [CronScheduleUpdate] -> Encoding

#omitField :: CronScheduleUpdate -> Bool

Eq CronScheduleUpdate Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#(==) :: CronScheduleUpdate -> CronScheduleUpdate -> Bool

#(/=) :: CronScheduleUpdate -> CronScheduleUpdate -> Bool

Generic CronScheduleUpdate Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Associated Types

type Rep CronScheduleUpdate

Instance details

Defined in Arbiter.Core.CronSchedule

type Rep CronScheduleUpdate = D1 ('MetaData "CronScheduleUpdate" "Arbiter.Core.CronSchedule" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "CronScheduleUpdate" 'PrefixI 'True) ((S1 ('MetaSel ('Just "overrideExpression") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Text))) :*: S1 ('MetaSel ('Just "overrideOverlap") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Text)))) :*: (S1 ('MetaSel ('Just "overrideTimezone") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Text))) :*: S1 ('MetaSel ('Just "enabled") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Bool)))))

Methods

#from :: CronScheduleUpdate -> Rep CronScheduleUpdate x

#to :: Rep CronScheduleUpdate x -> CronScheduleUpdate

Show CronScheduleUpdate Source #

Instance details

Defined in Arbiter.Core.CronSchedule

Methods

#showsPrec :: Int -> CronScheduleUpdate -> ShowS

#show :: CronScheduleUpdate -> String

#showList :: [CronScheduleUpdate] -> ShowS

type Rep CronScheduleUpdate Source #

Instance details

Defined in Arbiter.Core.CronSchedule

type Rep CronScheduleUpdate = D1 ('MetaData "CronScheduleUpdate" "Arbiter.Core.CronSchedule" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "CronScheduleUpdate" 'PrefixI 'True) ((S1 ('MetaSel ('Just "overrideExpression") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Text))) :*: S1 ('MetaSel ('Just "overrideOverlap") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Text)))) :*: (S1 ('MetaSel ('Just "overrideTimezone") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Text))) :*: S1 ('MetaSel ('Just "enabled") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Bool)))))

Source #effectiveExpression :: CronScheduleRow -> Text

The override expression if set, else the default.

Source #effectiveOverlap :: CronScheduleRow -> Text

The override overlap policy if set, else the default.

Source #effectiveTimezone :: CronScheduleRow -> Maybe Text

The override timezone if set, else the default. Nothing means UTC.

Source #data WorkerHealth

Heartbeat-derived health of a worker. Independent of its paused flag.

Constructors

Live
Stale
Draining

Instances

Instances details

FromJSON WorkerHealth Source #

Instance details

Defined in Arbiter.Core.Worker

Methods

#parseJSON :: Value -> Parser WorkerHealth

#parseJSONList :: Value -> Parser [WorkerHealth]

#omittedField :: Maybe WorkerHealth

ToJSON WorkerHealth Source #

Instance details

Defined in Arbiter.Core.Worker

Methods

#toJSON :: WorkerHealth -> Value

#toEncoding :: WorkerHealth -> Encoding

#toJSONList :: [WorkerHealth] -> Value

#toEncodingList :: [WorkerHealth] -> Encoding

#omitField :: WorkerHealth -> Bool

Eq WorkerHealth Source #

Instance details

Defined in Arbiter.Core.Worker

Methods

#(==) :: WorkerHealth -> WorkerHealth -> Bool

#(/=) :: WorkerHealth -> WorkerHealth -> Bool

Generic WorkerHealth Source #

Instance details

Defined in Arbiter.Core.Worker

Associated Types

type Rep WorkerHealth
Instance details Defined in Arbiter.Core.Worker type Rep WorkerHealth = D1 ('MetaData "WorkerHealth" "Arbiter.Core.Worker" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "Live" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "Stale" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "Draining" 'PrefixI 'False) (U1 :: Type -> Type)))

Methods

#from :: WorkerHealth -> Rep WorkerHealth x

#to :: Rep WorkerHealth x -> WorkerHealth

Show WorkerHealth Source #

Instance details

Defined in Arbiter.Core.Worker

Methods

#showsPrec :: Int -> WorkerHealth -> ShowS

#show :: WorkerHealth -> String

#showList :: [WorkerHealth] -> ShowS

type Rep WorkerHealth Source #

Instance details

Defined in Arbiter.Core.Worker

type Rep WorkerHealth = D1 ('MetaData "WorkerHealth" "Arbiter.Core.Worker" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "Live" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "Stale" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "Draining" 'PrefixI 'False) (U1 :: Type -> Type)))

Source #workerHealthFromText :: Text -> Either Text WorkerHealth

Decode the health SQL token into a WorkerHealth.

Connection pool settings

module Arbiter.Core.PoolConfig

Listener types

Notification collides with Database.PostgreSQL.Simple.Notification. Import it from Arbiter.Core.Listen.

Source #data Listener

Everything the worker needs to run a shared listener for one env.

Source #data RunningHub

Mutable hub state behind the slot. Opaque to callers.

Source #data HubLog

How a registrant wants hub events reported.

Constructors

HubLog
Fields hubRecovered :: Text -> IO () The hub's connection came back. hubWarn :: Text -> IO () This registrant's own channel handler raised. hubError :: Text -> IO () The hub's connection failed. hubRepeatInterval :: NominalDiffTime How often a standing connection failure says so again.

Source #withChannels :: MonadUnliftIO m => Listener -> HubLog -> [(ByteString, Notification -> m ())] -> (STM Bool -> m a) -> m a

Register this pool's channels on the env's shared hub for the duration of k. The STM action reports True once they are all subscribed.

Source #newPoolListener :: ((Connection -> IO ()) -> IO ()) -> IO Listener

Build a pool-backed Listener with a fresh hub slot from a connection runner.

Source #data DedicatedListen

A listener over its own libpq connection opened from a connection string.

Source #newDedicatedListen :: MonadIO m => ByteString -> m DedicatedListen

Allocate a DedicatedListen from a connection string, once at startup.

Source #dedicatedListener :: DedicatedListen -> Listener

The Listener for a DedicatedListen.