postgres

package
v0.18.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: Apache-2.0 Imports: 37 Imported by: 0

Documentation

Overview

Copyright IBM Corp. All Rights Reserved. SPDX-License-Identifier: Apache-2.0

Copyright IBM Corp. All Rights Reserved. SPDX-License-Identifier: Apache-2.0

Index

Constants

This section is empty.

Variables

View Source 
var AllOperations = []driver.Operation{driver.Insert, driver.Update, driver.Delete}

Functions

func NewAdvisoryLockFactory 

func NewAdvisoryLockFactory() func(context.Context, *sql.DB, int64) (tokensdriver.RecoveryLeadership, bool, error)

NewAdvisoryLockFactory returns a recovery leader factory function that uses PostgreSQL advisory locks.

func NewCleanupLeaderFactory 

func NewCleanupLeaderFactory() func(context.Context, *sql.DB, int64) (tokensdriver.CleanupLeadership, bool, error)

NewCleanupLeaderFactory returns a cleanup leader factory function that uses PostgreSQL advisory locks.

func NewConditionInterpreter 

func NewConditionInterpreter() *interpreter

func NewNamedDriver 

func NewNamedDriver(config driver3.Config, dbProvider fscPostgres.DbProvider) driver3.NamedDriver

NewNamedDriver returns a NamedDriver for Postgres.

func NewPaginationInterpreter 

func NewPaginationInterpreter() common.PagInterpreter

Types

type AdvisoryLock 

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

AdvisoryLock implements RecoveryLeadership using PostgreSQL advisory locks. Advisory locks are session-scoped and automatically released when the connection closes.

func NewAdvisoryLock 

func NewAdvisoryLock(ctx context.Context, db *sql.DB, lockID int64) (*AdvisoryLock, bool, error)

NewAdvisoryLock attempts to acquire a PostgreSQL advisory lock for the given lockID. Returns (lock, true, nil) if the lock was acquired successfully. Returns (nil, false, nil) if the lock is held by another session. Returns (nil, false, error) if an error occurred during acquisition.

The lock is session-scoped and will be automatically released when: - Close() is called explicitly - The connection is closed - The process terminates

func (*AdvisoryLock) Close 

func (l *AdvisoryLock) Close() error

Close releases the advisory lock and closes the connection. It is safe to call Close multiple times.

type AuditTransactionStore 

type AuditTransactionStore struct {
	*sqlcommon.TransactionStore
	// contains filtered or unexported fields
}

AuditTransactionStore wraps common.TransactionStore to add advisory lock to schema creation

func NewAuditTransactionStore 

func NewAuditTransactionStore(dbs *scommon.RWDB, tableNames sqlcommon.TableNames) (*AuditTransactionStore, error)

NewAuditTransactionStore creates a new AuditTransactionStore.

func (*AuditTransactionStore) CreateSchema 

func (s *AuditTransactionStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*AuditTransactionStore) GetSchema 

func (s *AuditTransactionStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

func (*AuditTransactionStore) WriteDB 

func (s *AuditTransactionStore) WriteDB() *sql.DB

WriteDB returns the underlying write *sql.DB. Used by the auditor distributed locker to share the connection pool.

type Driver 

type Driver struct {

	// Lazy providers for various store types to ensure they are initialized only when needed.
	TokenLock lazy.Provider[fscPostgres.Config, *TokenLockStore]
	Wallet    lazy.Provider[fscPostgres.Config, *WalletStore]
	Identity  lazy.Provider[fscPostgres.Config, *IdentityStore]
	Token     lazy.Provider[fscPostgres.Config, *TokenStore]
	AuditTx   lazy.Provider[fscPostgres.Config, *AuditTransactionStore]
	OwnerTx   lazy.Provider[fscPostgres.Config, *TransactionStore]
	Endorser  lazy.Provider[fscPostgres.Config, *EndorserStore]
	KeyStore  lazy.Provider[fscPostgres.Config, *KeystoreStore]
	// contains filtered or unexported fields
}

Driver implements the token storage driver for Postgres.

func NewDriver 

func NewDriver(config driver3.Config) *Driver

NewDriver returns a new Driver for Postgres using the default database provider.

func NewDriverWithDbProvider 

func NewDriverWithDbProvider(config driver3.Config, dbProvider fscPostgres.DbProvider) *Driver

NewDriverWithDbProvider returns a new Driver for Postgres using the given database provider.

func (*Driver) NewAuditTransaction 

func (d *Driver) NewAuditTransaction(name driver2.PersistenceName, params ...string) (driver3.AuditTransactionStore, error)

NewAuditTransaction returns a new AuditTransactionStore.

func (*Driver) NewEndorser 

func (d *Driver) NewEndorser(name driver2.PersistenceName, params ...string) (driver3.EndorserStore, error)

NewEndorser returns a new EndorserStore.

func (*Driver) NewIdentity 

func (d *Driver) NewIdentity(name driver2.PersistenceName, params ...string) (driver3.IdentityStore, error)

NewIdentity returns a new IdentityStore.

func (*Driver) NewKeyStore 

func (d *Driver) NewKeyStore(name driver2.PersistenceName, params ...string) (driver3.KeyStore, error)

NewKeyStore returns a new KeyStoreStore.

func (*Driver) NewOwnerTransaction 

func (d *Driver) NewOwnerTransaction(name driver2.PersistenceName, params ...string) (driver3.TokenTransactionStore, error)

NewOwnerTransaction returns a new TokenTransactionStore.

func (*Driver) NewToken 

func (d *Driver) NewToken(name driver2.PersistenceName, params ...string) (driver3.TokenStore, error)

NewToken returns a new TokenStore.

func (*Driver) NewTokenLock 

func (d *Driver) NewTokenLock(name driver2.PersistenceName, params ...string) (driver3.TokenLockStore, error)

NewTokenLock returns a new TokenLockStore.

func (*Driver) NewWallet 

func (d *Driver) NewWallet(name driver2.PersistenceName, params ...string) (driver3.WalletStore, error)

NewWallet returns a new WalletStore.

type EndorserStore 

type EndorserStore struct {
	*common3.EndorserStore
	// contains filtered or unexported fields
}

EndorserStore wraps common.EndorserStore to add advisory lock to schema creation

func NewEndorserStore 

func NewEndorserStore(dbs *scommon.RWDB, tables common3.TableNames) (*EndorserStore, error)

NewEndorserStore creates a new EndorserStore for Postgres

func (*EndorserStore) CreateSchema 

func (s *EndorserStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*EndorserStore) GetSchema 

func (s *EndorserStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

type IdentityNotifier 

type IdentityNotifier struct {
	*Notifier
}

IdentityNotifier handles notifications for identity configurations.

func NewIdentityNotifier 

func NewIdentityNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, dataSource string) (*IdentityNotifier, error)

NewIdentityNotifier returns a new IdentityNotifier for the given RWDB and table names.

func (*IdentityNotifier) Subscribe 

func (n *IdentityNotifier) Subscribe(callback func(idriver.Operation, idriver.IdentityConfigurationRecord)) error

Subscribe registers a callback function to be called when an identity configuration is inserted or updated.

type IdentityStore 

type IdentityStore struct {
	*sqlcommon.IdentityStore
	// contains filtered or unexported fields
}

IdentityStore wraps common.IdentityStore to add advisory lock to schema creation

func NewIdentityStore 

func NewIdentityStore(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, dataSource string) (*IdentityStore, error)

NewIdentityStore creates a new IdentityStore with advisory lock support

func (*IdentityStore) CreateSchema 

func (s *IdentityStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*IdentityStore) GetSchema 

func (s *IdentityStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

type KeystoreStore 

type KeystoreStore struct {
	*common3.KeystoreStore
	// contains filtered or unexported fields
}

KeystoreStore wraps common.KeystoreStore to add advisory lock to schema creation

func NewKeystoreStore 

func NewKeystoreStore(dbs *common2.RWDB, tableNames common3.TableNames) (*KeystoreStore, error)

NewKeystoreStore returns a new KeystoreStore for the given RWDB and table names.

func (*KeystoreStore) CreateSchema 

func (s *KeystoreStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*KeystoreStore) GetSchema 

func (s *KeystoreStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

type Notifier 

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

Notifier implements a simple subscription API to listen for updates on a database table.

func NewNotifier 

func NewNotifier(
	writeDB *sql.DB,
	table, dataSource string,
	notifyOperations []driver.Operation,
	primaryKeys ...PrimaryKey,
) *Notifier

NewNotifier returns a new Notifier for the given RWDB and table names.

func (*Notifier) Close 

func (db *Notifier) Close() error

Close stops the listener and cleans up resources.

func (*Notifier) CreateSchema 

func (db *Notifier) CreateSchema() error

CreateSchema creates the notification objects in the database. It returns an error if the schema creation fails.

func (*Notifier) GetSchema 

func (db *Notifier) GetSchema() string

GetSchema returns the SQL schema for creating the notification objects in the database.

func (*Notifier) ListenerError 

func (db *Notifier) ListenerError() <-chan error

ListenerError returns a channel that receives errors from the listener. The caller should consume this channel to detect listener failures.

func (*Notifier) Subscribe 

func (db *Notifier) Subscribe(callback driver.TriggerCallback) error

Subscribe registers a callback function to be called when a matching database event occurs. It returns an error if the notifier is closed or if the listener fails to start.

func (*Notifier) TransportError added in v0.18.0 

func (db *Notifier) TransportError() <-chan error

TransportError returns a channel that receives non-fatal transport errors reported by the underlying listener (e.g. a dropped LISTEN connection that is being reconnected). The notifier keeps running across such errors; the channel merely lets a consumer learn that notifications may have been missed during the outage. Sends are best-effort and buffered depth 1, so only the most recent error is retained when the channel is not being drained.

func (*Notifier) UnsubscribeAll 

func (db *Notifier) UnsubscribeAll() error

UnsubscribeAll removes all subscribers.

type PrimaryKey 

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

PrimaryKey represents a primary key column with its value decoder

func NewBytePrimaryKey 

func NewBytePrimaryKey(name driver.ColumnKey) *PrimaryKey

func NewSimplePrimaryKey 

func NewSimplePrimaryKey(name driver.ColumnKey) *PrimaryKey

type TokenLockStore 

type TokenLockStore struct {
	*common5.TokenLockStore
	// contains filtered or unexported fields
}

TokenLockStore implements the token lock storage for Postgres.

func NewTokenLockStore 

func NewTokenLockStore(dbs *common2.RWDB, tableNames common5.TableNames) (*TokenLockStore, error)

NewTokenLockStore returns a new TokenLockStore for the given RWDB and table names.

func (*TokenLockStore) AcquireCleanupLeadership added in v0.17.0 

func (db *TokenLockStore) AcquireCleanupLeadership(ctx context.Context) (driver.CleanupLeadership, bool, error)

AcquireCleanupLeadership attempts to acquire a Postgres advisory lock so only one replica runs Cleanup per tick for this TMS; others skip the tick and release no held resources, so contention is limited to the acquire attempt itself. The lock id and factory are fixed at construction. Note the winner holds a dedicated connection off writeDB for the tick's duration (see NewAdvisoryLock), so writeDB's connection pool needs at least one spare connection beyond normal write traffic. See #1798.

func (*TokenLockStore) Cleanup 

func (db *TokenLockStore) Cleanup(ctx context.Context, leaseExpiry time.Duration) error

Cleanup removes stale token locks that have expired. The deletion itself is the backend-independent one implemented by the embedded store; Postgres only adds the logging of the rows that are about to go.

func (*TokenLockStore) CreateSchema 

func (s *TokenLockStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*TokenLockStore) GetSchema 

func (s *TokenLockStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

type TokenNotifier 

type TokenNotifier struct {
	*Notifier
}

TokenNotifier handles notifications for tokens.

func NewTokenNotifier 

func NewTokenNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, dataSource string) (*TokenNotifier, error)

NewTokenNotifier returns a new TokenNotifier for the given RWDB and table names.

func (*TokenNotifier) Subscribe 

func (n *TokenNotifier) Subscribe(callback func(tokensdriver.Operation, tokensdriver.TokenRecordReference)) error

Subscribe registers a callback function to be called when a token is inserted, updated, or deleted.

type TokenStore 

type TokenStore struct {
	*sqlcommon.TokenStore
	// contains filtered or unexported fields
}

TokenStore wraps common.TokenStore to add advisory lock to schema creation

func NewTokenStoreWithNotifier 

func NewTokenStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, notifier *TokenNotifier) (*TokenStore, error)

func (*TokenStore) CreateSchema 

func (s *TokenStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*TokenStore) GetSchema 

func (s *TokenStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

type TransactionNotifier 

type TransactionNotifier struct {
	*Notifier
}

TransactionNotifier handles notifications for transaction status changes.

func NewTransactionNotifier 

func NewTransactionNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, dataSource string) (*TransactionNotifier, error)

NewTransactionNotifier returns a new TransactionNotifier for the given RWDB and table names.

func (*TransactionNotifier) Subscribe 

func (n *TransactionNotifier) Subscribe(callback func(tokensdriver.Operation, tokensdriver.TransactionRecordReference)) error

Subscribe registers a callback function to be called when a transaction request status is updated.

type TransactionStore 

type TransactionStore struct {
	*sqlcommon.TransactionStore
	// contains filtered or unexported fields
}

TransactionStore extends the common TransactionStore with PostgreSQL-specific atomic claim operations.

func NewTransactionStoreWithNotifier 

func NewTransactionStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, notifier *TransactionNotifier) (*TransactionStore, error)

NewTransactionStoreWithNotifier creates a new TransactionStore with the provided notifier and recovery support.

func (*TransactionStore) ClaimPendingTransactions 

func (db *TransactionStore) ClaimPendingTransactions(ctx context.Context, params tokensdriver.RecoveryClaimParams) ([]*tokensdriver.RecoveryClaim, error)

ClaimPendingTransactions atomically claims a batch of pending transactions using PostgreSQL's UPDATE...RETURNING. This ensures only one recovery instance can claim a specific transaction. All state we need lives on the requests table (tx_id PK + stored_at + status + recovery_claim_* lease columns); the transactions table is no longer touched. RETURNING tx_id, stored_at directly from the UPDATE removes the outer join the previous CTE used to recover the timestamp.

func (*TransactionStore) CleanupExpiredClaims 

func (db *TransactionStore) CleanupExpiredClaims(ctx context.Context) (int, error)

CleanupExpiredClaims removes expired recovery claims. Returns the number of claims cleaned up.

func (*TransactionStore) CreateSchema 

func (s *TransactionStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*TransactionStore) GetSchema 

func (s *TransactionStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

func (*TransactionStore) ReleaseRecoveryClaim 

func (db *TransactionStore) ReleaseRecoveryClaim(ctx context.Context, txID string, owner string, message string) error

ReleaseRecoveryClaim releases the recovery claim on a transaction. This clears the claim metadata and optionally updates the status message.

type WalletStore 

type WalletStore struct {
	*common3.WalletStore
	// contains filtered or unexported fields
}

WalletStore wraps common.WalletStore to add advisory lock to schema creation

func NewWalletStore 

func NewWalletStore(dbs *common2.RWDB, tableNames common3.TableNames) (*WalletStore, error)

NewWalletStore returns a new WalletStore for the given RWDB and table names.

func (*WalletStore) CreateSchema 

func (s *WalletStore) CreateSchema() error

CreateSchema overrides the base CreateSchema to ensure GetSchema is called on the correct receiver

func (*WalletStore) GetSchema 

func (s *WalletStore) GetSchema() string

GetSchema overrides the base GetSchema to prefix with advisory lock

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.