func NewPgAdvisoryLock(groupLockID int64, connStr string) (*PgAdvisoryLock, error) {
	lock := &PgAdvisoryLock{
		connStr:      connStr,
		groupLockID:  groupLockID,
		afterConnect: checkConnection,
	}
	return lock, nil
}
NewPgAdvisoryLock creates a new instance with specified lock ID, connection pool and lock timeout.
package util
import (
	"context"
	"fmt"
	"strconv"
	"sync"
	"time"
	pgx "github.com/jackc/pgx/v4"
	"github.com/timescale/promscale/pkg/log"
)
const defaultConnectionTimeout = time.Minute
var SharedLeaseFailure = fmt.Errorf("failed to acquire shared lease")
func GetSharedLease(ctx context.Context, conn *pgx.Conn, id int64) error {
	gotten, err := runLockFunction(ctx, conn, "SELECT pg_try_advisory_lock_shared($1)", id)
	if err != nil {
		return fmt.Errorf("Unable to get shared schema lock. Please make sure that no operations requiring exclusive locking are running: %w", err)
	}
	if !gotten {
		return SharedLeaseFailure
	}
	return nil
}
type PgLeaderLock struct {
	PgAdvisoryLock
	obtained        bool
	numLockAttempts int
}
type AfterConnectFunc = func(ctx context.Context, conn *pgx.Conn) error
func NewPgLeaderLock(groupLockID int64, connStr string, afterConnect AfterConnectFunc) (*PgLeaderLock, error) {
	if afterConnect == nil {
		afterConnect = checkConnection
	}
	lock := &PgLeaderLock{
		PgAdvisoryLock{
			connStr:      connStr,
			groupLockID:  groupLockID,
			afterConnect: afterConnect,
		},
		false,
		0,
	}
	_, err := lock.tryLock()
	if err != nil {
		return nil, err
	}
	return lock, nil
}
func (l *PgLeaderLock) ID() string {
	return strconv.FormatInt(int64(l.groupLockID), 10)
}
func (l *PgLeaderLock) BecomeLeader() (bool, error) {
	return l.tryLock()
}
func (l *PgLeaderLock) IsLeader() (bool, error) {
	return l.tryLock()
}
func (l *PgLeaderLock) Resign() error {
	log.Info("msg", "Resigning as the leader (will no longer be writing)", "component", "leader_election", "status", "follower", "group_id", l.groupLockID)
	return l.release()
}
func (l *PgLeaderLock) tryLock() (bool, error) {
	l.mutex.Lock()
	defer l.mutex.Unlock()
	defer func() { l.numLockAttempts++ }()
	if l.obtained && l.conn != nil {
		err := l.conn.QueryRow(context.Background(), "SELECT").Scan()
		if err != nil {
			l.obtained = false
			l.connCleanUp()
			return false, err
		}
		return true, nil
	}
	gotLock, err := l.getAdvisoryLock()
	if err != nil {
		l.obtained = false
		l.connCleanUp()
		return false, err
	}
	if !gotLock {
		if l.numLockAttempts == 0 {
			log.Info("msg", "I am starting as a follower (will not be writing until I become the leader)", "component", "leader_election", "status", "follower", "group_id", l.groupLockID)
		} else if l.obtained {
			log.Info("msg", "I have lost the leader lock and am now a follower (will not be writing)", "component", "leader_election", "status", "follower", "group_id", l.groupLockID)
		}
		l.obtained = false
		return false, nil
	}
	if !l.obtained {
		l.obtained = true
		log.Info("msg", "I have become the leader (starting to write incoming data)", "component", "leader_election", "status", "leader", "group_id", l.groupLockID)
	}
	return true, nil
}
func (l *PgLeaderLock) locked() bool {
	l.mutex.RLock()
	defer l.mutex.RUnlock()
	return l.obtained
}
func (l *PgLeaderLock) release() error {
	l.mutex.Lock()
	defer l.mutex.Unlock()
	if !l.obtained {
		return fmt.Errorf("can't release while not holding the lock")
	}
	defer func() { l.obtained = false }()
	unlocked, err := l.unlock()
	if err != nil {
		l.connCleanUp()
		return err
	}
	if !unlocked {
		log.Debug("msg", fmt.Sprintf("release for a lock that was not held: group id %d", l.groupLockID))
	}
	return nil
}
type PgAdvisoryLock struct {
	conn         *pgx.Conn
	connStr      string
	groupLockID  int64
	afterConnect AfterConnectFunc
	mutex sync.RWMutex
}
type AdvisoryLock interface {
	GetAdvisoryLock() (bool, error)
	GetSharedAdvisoryLock() (bool, error)
	Unlock() (bool, error)
	UnlockShared() (bool, error)
	Close()
}
var _ AdvisoryLock = (*PgAdvisoryLock)(nil)