job

Documentation

Index

• func NewORM(db *gorm.DB, config Config, pipelineORM pipeline.ORM, ...) *orm
• func NewSpawner(orm ORM, config Config) *spawner
• type Config
• type Delegate
• type ORM
• type Service
• type Spawner
• type Spec
• type Type

Functions

func NewORM

func NewORM(db *gorm.DB, config Config, pipelineORM pipeline.ORM, eventBroadcaster postgres.EventBroadcaster, advisoryLocker postgres.AdvisoryLocker) *orm

func NewSpawner

func NewSpawner(orm ORM, config Config) *spawner

Types

type Config

type Config interface {
	DatabaseMaximumTxDuration() time.Duration
	DatabaseURL() string
	JobPipelineDBPollInterval() time.Duration
	JobPipelineParallelism() uint8
}

type Delegate

type Delegate interface {
	JobType() Type
	ToDBRow(spec Spec) models.JobSpecV2
	FromDBRow(spec models.JobSpecV2) Spec
	ServicesForSpec(spec Spec) ([]Service, error)
}

TODO(spook): I can't wait for Go generics

type ORM

type ORM interface {
	ListenForNewJobs() (postgres.Subscription, error)
	ListenForDeletedJobs() (postgres.Subscription, error)
	ClaimUnclaimedJobs(ctx context.Context) ([]models.JobSpecV2, error)
	CreateJob(ctx context.Context, jobSpec *models.JobSpecV2, taskDAG pipeline.TaskDAG) error
	DeleteJob(ctx context.Context, id int32) error
	RecordError(ctx context.Context, jobID int32, description string)
	UnclaimJob(ctx context.Context, id int32) error
	CheckForDeletedJobs(ctx context.Context) (deletedJobIDs []int32, err error)
	Close() error
}

type Service

type Service interface {
	Start() error
	Close() error
}

type Spawner

type Spawner interface {
	Start()
	Stop()
	CreateJob(ctx context.Context, spec Spec) (int32, error)
	DeleteJob(ctx context.Context, jobID int32) error
	RegisterDelegate(delegate Delegate)
}

The job spawner manages the spinning up and spinning down of the long-running services that perform the work described by job specs. Each active job spec has 1 or more of these services associated with it.

At present, Flux Monitor and Offchain Reporting jobs can only have a single "initiator", meaning that they only require a single service. But the older "direct request" model allows for multiple initiators, which imply multiple services.

type Spec

type Spec interface {
	JobID() int32
	JobType() Type
	TaskDAG() pipeline.TaskDAG
	TableName() string
}

type Type

type Type string