const (
pgConnRecycleSeconds = 1800 // applies for monitored nodes
applicationName = "pgwatch" // will be set on all opened PG connections for informative purposes
)
Deadline defaults for the various database round-trips pgwatch performs.
These are package-level vars rather than consts so tests can shrink them (e.g. to a few milliseconds) to exercise fault-injection paths quickly.
`internal/db` is a leaf package — it MUST NOT import `internal/reaper` or `internal/sources`. Tests of the call sites shrink these vars locally and restore them on cleanup.
var (
// MinFetchTimeout is the floor applied to fetch-path deadlines. Very
// short metric intervals still get a usable window to dial, authenticate,
// and return a meaningful error rather than aborting before the round-trip
// starts.
MinFetchTimeout = 30 * time.Second
// ChangeDetectionTimeout bounds the Detect*Changes family and the
// QueryMeasurements helper used by them.
ChangeDetectionTimeout = 60 * time.Second
// RuntimeInfoTimeout bounds each sub-query of FetchRuntimeInfo:
// version, platform discovery, approximate size, extensions, available
// extensions.
RuntimeInfoTimeout = 30 * time.Second
// ResolverTimeout bounds a single ResolveDatabasesFromPostgres call
// (pool creation + discovery query).
ResolverTimeout = 15 * time.Second
// PingTimeoutMargin is added on top of the configured ConnectTimeout when
// bounding the main-loop Ping gate. Default 5s + 5s = 10s total.
PingTimeoutMargin = 5 * time.Second
)
func DoesSchemaExist(ctx context.Context, conn PgxIface, schema string) (bool, error)
DoesSchemaExist checks if schema exists
func Init(ctx context.Context, db PgxPoolIface, init ConnInitCallback) error
Init checks if connection is establised. If not, retries connection 3 times with delay 1s
func IsClientOnSameHost(conn PgxIface) (bool, error)
Function to determine if the client is connected to the same host as the PostgreSQL server
func IsPgConnStr(arg string) bool
func MarshallParamToJSONB(v any) any
func NeedsMigration(storage any, needsMigrationErr error) error
func Ping(ctx context.Context, connStr string) error
func WithFetchTimeout(ctx context.Context, op string, interval time.Duration) (context.Context, context.CancelFunc)
WithFetchTimeout returns a child of ctx whose deadline is max(interval, MinFetchTimeout). The op string is embedded in the context's cause so a context.DeadlineExceeded is distinguishable per call site.
The cancel func MUST be called by the caller to release the timer once the work completes (or fails) — same convention as context.WithTimeout.
func WithOpTimeout(ctx context.Context, op string, d time.Duration) (context.Context, context.CancelFunc)
WithOpTimeout returns a child of ctx with a fixed deadline of d. The op string is embedded in the context's cause so a context.DeadlineExceeded is distinguishable per call site.
The cancel func MUST be called by the caller to release the timer.
func deadlineCause(op string) error
deadlineCause builds the canonical cause string attached to a derived context via context.WithTimeoutCause. The format "<op> deadline" is greppable and distinct per call site.
type ConnConfigCallback = func(*pgxpool.Config) error
type ConnInitCallback = func(context.Context, PgxIface) error
Migrator is an interface for database schema migration
type Migrator interface {
Migrate() error
NeedsMigration() (bool, error)
}
PgxConnIface is interface representing pgx connection
type PgxConnIface interface {
PgxIface
BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
Close(ctx context.Context) error
Ping(ctx context.Context) error
}
PgxIface is common interface for every pgx class
type PgxIface interface {
Begin(ctx context.Context) (pgx.Tx, error)
Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
QueryRow(context.Context, string, ...any) pgx.Row
Query(ctx context.Context, query string, args ...any) (pgx.Rows, error)
CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
}
PgxPoolIface is interface representing pgx pool
type PgxPoolIface interface {
PgxIface
Acquire(ctx context.Context) (*pgxpool.Conn, error)
BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
Close()
Config() *pgxpool.Config
Ping(ctx context.Context) error
SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
Stat() *pgxpool.Stat
}
func New(ctx context.Context, connStr string, callbacks ...ConnConfigCallback) (PgxPoolIface, error)
New create a new pool
func NewWithConfig(ctx context.Context, connConfig *pgxpool.Config, callbacks ...ConnConfigCallback) (PgxPoolIface, error)
NewWithConfig creates a new pool with a given config
type Querier interface {
Query(ctx context.Context, query string, args ...any) (pgx.Rows, error)
}