postgres

Documentation

Index

• Constants
• Variables
• func DBWithDefaultContext(db *gorm.DB, fc func(db *gorm.DB) error) error
• func DefaultQueryCtx() (ctx context.Context, cancel context.CancelFunc)
• func GormTransaction(ctx context.Context, db *gorm.DB, fc func(tx *gorm.DB) error, ...) (err error)
• func GormTransactionWithDefaultContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) error
• func GormTransactionWithoutContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) (err error)
• func IsSerializationAnomaly(err error) bool
• func MustSQLDB(db *gorm.DB) *sql.DB
• func MustSQLTx(db *gorm.DB) *sql.Tx
• func NewEventBroadcaster(uri url.URL, minReconnectInterval time.Duration, ...) *eventBroadcaster
• type AdvisoryLocker
  • func NewAdvisoryLock(uri url.URL) AdvisoryLocker
• type Event
• type EventBroadcaster
• type NullAdvisoryLocker
  • func (n *NullAdvisoryLocker) Close() error
  • func (*NullAdvisoryLocker) Unlock(ctx context.Context, classID int32, objectID int32) error
  • func (*NullAdvisoryLocker) WithAdvisoryLock(ctx context.Context, classID int32, objectID int32, f func() error) error
• type NullEventBroadcaster
  • func (*NullEventBroadcaster) Close() error
  • func (*NullEventBroadcaster) Healthy() error
  • func (*NullEventBroadcaster) Notify(channel string, payload string) error
  • func (*NullEventBroadcaster) NotifyInsideGormTx(tx *gorm.DB, channel string, payload string) error
  • func (*NullEventBroadcaster) Ready() error
  • func (*NullEventBroadcaster) Start() error
  • func (*NullEventBroadcaster) Subscribe(channel, payloadFilter string) (Subscription, error)
• type Subscription

Constants

const (
	AdvisoryLockClassID_EthBroadcaster int32 = 0
	AdvisoryLockClassID_JobSpawner     int32 = 1
	AdvisoryLockClassID_EthConfirmer   int32 = 2

	// ORM takes lock on 1027321974924625846 which splits into ClassID 239192036, ObjID 2840971190
	AdvisoryLockClassID_ORM int32 = 239192036

	AdvisoryLockObjectID_EthConfirmer int32 = 0
)

NOTE: All advisory lock class IDs used by the Chainlink application MUST be kept here to avoid accidental re-use

const (
	ChannelJobCreated   = "insert_on_jobs"
	ChannelJobDeleted   = "delete_from_jobs"
	ChannelRunStarted   = "pipeline_run_started"
	ChannelRunCompleted = "pipeline_run_completed"

	// Postgres channel to listen for new eth_txes
	ChannelInsertOnEthTx = "insert_on_eth_txes"
)

const (
	// LockTimeout controls the max time we will wait for any kind of database lock.
	// It's good to set this to _something_ because waiting for locks forever is really bad.
	LockTimeout = 15 * time.Second
	// IdleInTxSessionTimeout controls the max time we leave a transaction open and idle.
	// It's good to set this to _something_ because leaving transactions open forever is really bad.
	IdleInTxSessionTimeout = 1 * time.Hour
)

NOTE: In an ideal world the timeouts below would be set to something sane in the postgres configuration by the user. Since we do not live in an ideal world, it is necessary to override them here.

They cannot easily be set at a session level due to how Go's connection pooling works.

const DefaultQueryTimeout = 10 * time.Second

Variables

var (
	DefaultSqlTxOptions = sql.TxOptions{

		Isolation: sql.LevelReadCommitted,
	}
)

var (
	ErrNoDeadlineSet = errors.New("no deadline set")
)

Functions

func DBWithDefaultContext

func DBWithDefaultContext(db *gorm.DB, fc func(db *gorm.DB) error) error

func DefaultQueryCtx

func DefaultQueryCtx() (ctx context.Context, cancel context.CancelFunc)

DefaultQueryCtx returns a context with a sensible sanity limit timeout for SQL queries

func GormTransaction

func GormTransaction(ctx context.Context, db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) (err error)

func GormTransactionWithDefaultContext

func GormTransactionWithDefaultContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) error

func GormTransactionWithoutContext

func GormTransactionWithoutContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) (err error)

WARNING: Only use for nested txes inside ORM methods where you expect db to already have a ctx with a deadline.

func IsSerializationAnomaly

func IsSerializationAnomaly(err error) bool

func MustSQLDB

func MustSQLDB(db *gorm.DB) *sql.DB

func MustSQLTx

func MustSQLTx(db *gorm.DB) *sql.Tx

MustSQLDB panics if there is an error getting the underlying SQL TX

func NewEventBroadcaster

func NewEventBroadcaster(uri url.URL, minReconnectInterval time.Duration, maxReconnectDuration time.Duration) *eventBroadcaster

Types

type AdvisoryLocker

type AdvisoryLocker interface {
	Unlock(ctx context.Context, classID int32, objectID int32) error
	WithAdvisoryLock(ctx context.Context, classID int32, objectID int32, f func() error) error
	Close() error
}

func NewAdvisoryLock

func NewAdvisoryLock(uri url.URL) AdvisoryLocker

type Event

type Event struct {
	Channel string
	Payload string
}

type EventBroadcaster

type EventBroadcaster interface {
	service.Service
	Subscribe(channel, payloadFilter string) (Subscription, error)
	Notify(channel string, payload string) error
	NotifyInsideGormTx(tx *gorm.DB, channel string, payload string) error
}

EventBroadcaster opaquely manages a collection of Postgres event listeners and broadcasts events to subscribers (with an optional payload filter).

type NullAdvisoryLocker

type NullAdvisoryLocker struct {
	// contains filtered or unexported fields
}

func (*NullAdvisoryLocker) Close

func (n *NullAdvisoryLocker) Close() error

func (*NullAdvisoryLocker) Unlock

func (*NullAdvisoryLocker) Unlock(ctx context.Context, classID int32, objectID int32) error

func (*NullAdvisoryLocker) WithAdvisoryLock

func (*NullAdvisoryLocker) WithAdvisoryLock(ctx context.Context, classID int32, objectID int32, f func() error) error

type NullEventBroadcaster

type NullEventBroadcaster struct{}

NullEventBroadcaster implements null pattern for event broadcaster

func (*NullEventBroadcaster) Close

func (*NullEventBroadcaster) Close() error

func (*NullEventBroadcaster) Healthy

func (*NullEventBroadcaster) Healthy() error

func (*NullEventBroadcaster) Notify

func (*NullEventBroadcaster) Notify(channel string, payload string) error

func (*NullEventBroadcaster) NotifyInsideGormTx

func (*NullEventBroadcaster) NotifyInsideGormTx(tx *gorm.DB, channel string, payload string) error

func (*NullEventBroadcaster) Ready

func (*NullEventBroadcaster) Ready() error

func (*NullEventBroadcaster) Start

func (*NullEventBroadcaster) Start() error

func (*NullEventBroadcaster) Subscribe

func (*NullEventBroadcaster) Subscribe(channel, payloadFilter string) (Subscription, error)

type Subscription

type Subscription interface {
	Events() <-chan Event
	Close()

	ChannelName() string
	InterestedIn(event Event) bool
	Send(event Event)
}

Subscription represents a subscription to a Postgres event channel