diff --git a/go.mod b/go.mod index 6478cbed..cacd560b 100644 --- a/go.mod +++ b/go.mod @@ -5,8 +5,10 @@ go 1.26.0 require ( github.com/aws/aws-sdk-go v1.55.8 github.com/cloudflare/cloudflare-go v0.117.0 + github.com/fergusstrange/embedded-postgres v1.34.0 github.com/google/go-github v17.0.0+incompatible github.com/hashicorp/golang-lru/v2 v2.0.7 + github.com/jackc/pgx/v5 v5.10.0 github.com/mattn/go-sqlite3 v1.14.47 github.com/shopspring/decimal v1.4.0 go.sia.tech/core v0.21.4 @@ -27,18 +29,24 @@ require ( github.com/dunglas/httpsfv v1.1.0 // indirect github.com/goccy/go-json v0.10.5 // indirect github.com/google/go-querystring v1.1.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/jmespath/go-jmespath v0.4.0 // indirect github.com/julienschmidt/httprouter v1.3.0 // indirect + github.com/lib/pq v1.10.9 // indirect github.com/quic-go/qpack v0.6.0 // indirect github.com/quic-go/quic-go v0.60.0 // indirect github.com/quic-go/webtransport-go v0.11.0 // indirect github.com/rogpeppe/go-internal v1.12.0 // indirect + github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect go.etcd.io/bbolt v1.5.0 // indirect go.sia.tech/mux v1.5.2 // indirect go.sia.tech/web v0.0.0-20240610131903-5611d44a533e // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/crypto v0.53.0 // indirect golang.org/x/net v0.56.0 // indirect + golang.org/x/sync v0.21.0 // indirect golang.org/x/text v0.38.0 // indirect golang.org/x/tools v0.47.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect diff --git a/go.sum b/go.sum index f97dedee..8f72ef14 100644 --- a/go.sum +++ b/go.sum @@ -7,6 +7,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dunglas/httpsfv v1.1.0 h1:Jw76nAyKWKZKFrpMMcL76y35tOpYHqQPzHQiwDvpe54= github.com/dunglas/httpsfv v1.1.0/go.mod h1:zID2mqw9mFsnt7YC3vYQ9/cjq30q41W+1AnDwH8TiMg= +github.com/fergusstrange/embedded-postgres v1.34.0 h1:c6RKhPKFsLVU+Tdxsx8q0UxCHsvZZ/iShAnljRBXs6s= +github.com/fergusstrange/embedded-postgres v1.34.0/go.mod h1:w0YvnCgf19o6tskInrOOACtnqfVlOvluz3hlNLY7tRk= github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= @@ -18,6 +20,14 @@ github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg= github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo= github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8= @@ -28,6 +38,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= +github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/mattn/go-sqlite3 v1.14.47 h1:jOBI62gS7nKeZv+as1oGEy0+1qISgXwH/QBlR6KbfIo= github.com/mattn/go-sqlite3 v1.14.47/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -45,8 +57,12 @@ github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99 github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 h1:nIPpBwaJSVYIxUFsDv3M8ofmx9yWTog9BfvIu0q41lo= +github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8/go.mod h1:HUYIGzjTL3rfEspMxjDjgmT5uz5wzYJKVo23qUhYTos= go.etcd.io/bbolt v1.5.0 h1:S7GAl7Fxv12yohbwFfIbQCGDWbQbtDGPET4P/bD4lxU= go.etcd.io/bbolt v1.5.0/go.mod h1:mkltfYE5aUHQxUct9N9V+Kp7aSjFqjgrhcXIS70Lrdk= go.sia.tech/core v0.21.4 h1:EFbw8tJ2Jo4cqU1e6VNTUUg5WPGaUlpGD5t1RYje59c= @@ -96,6 +112,7 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EV gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= lukechampine.com/flagg v1.1.1 h1:jB5oL4D5zSUrzm5og6dDEi5pnrTF1poKfC7KE1lLsqc= diff --git a/persist/postgres/init.go b/persist/postgres/init.go new file mode 100644 index 00000000..1b5cfa57 --- /dev/null +++ b/persist/postgres/init.go @@ -0,0 +1,77 @@ +package postgres + +import ( + "context" + _ "embed" // for init.sql + "fmt" + "time" + + "go.sia.tech/core/types" + "go.uber.org/zap" + "lukechampine.com/frand" +) + +// init queries are run when the database is first created. +// +//go:embed init.sql +var initDatabase string + +func (s *Store) initNewDatabase(target int64) error { + return s.transaction(func(ctx context.Context, tx *txn) error { + if _, err := tx.Exec(ctx, initDatabase); err != nil { + return err + } else if err := setDBVersion(ctx, tx, target); err != nil { + return fmt.Errorf("failed to set initial database version: %w", err) + } else if err := generateHostKey(ctx, tx); err != nil { + return fmt.Errorf("failed to generate host key: %w", err) + } + return nil + }) +} + +func (s *Store) upgradeDatabase(current, target int64) error { + log := s.log.Named("migrations").With(zap.Int64("target", target)) + for ; current < target; current++ { + version := current + 1 // initial schema is version 1, migration 0 is version 2, etc. + log := log.With(zap.Int64("version", version)) + start := time.Now() + fn := migrations[current-1] + err := s.transaction(func(ctx context.Context, tx *txn) error { + if err := fn(ctx, tx, log); err != nil { + return err + } + return setDBVersion(ctx, tx, version) + }) + if err != nil { + return fmt.Errorf("migration %d failed: %w", version, err) + } + log.Info("migration complete", zap.Duration("elapsed", time.Since(start))) + } + return nil +} + +func (s *Store) init(target int64) error { + version := getDBVersion(context.Background(), s.pool) + switch { + case version == 0: + if err := s.initNewDatabase(target); err != nil { + return fmt.Errorf("failed to initialize database: %w", err) + } + case version < target: + s.log.Info("database version is out of date;", zap.Int64("version", version), zap.Int64("target", target)) + if err := s.upgradeDatabase(version, target); err != nil { + return fmt.Errorf("failed to upgrade database: %w", err) + } + case version > target: + return fmt.Errorf("database version %v is newer than expected %v. database downgrades are not supported", version, target) + } + // nothing to do + return nil +} + +func generateHostKey(ctx context.Context, tx *txn) (err error) { + key := types.NewPrivateKeyFromSeed(frand.Bytes(32)) + var dbID int64 + err = tx.QueryRow(ctx, `UPDATE global_settings SET host_key=$1 RETURNING id`, []byte(key)).Scan(&dbID) + return +} diff --git a/persist/postgres/init.sql b/persist/postgres/init.sql new file mode 100644 index 00000000..daf386f8 --- /dev/null +++ b/persist/postgres/init.sql @@ -0,0 +1,351 @@ +/* + When changing the schema, a new migration function must be added to + migrations.go +*/ + +CREATE TABLE wallet_siacoin_elements ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + siacoin_value NUMERIC(50,0) NOT NULL, + sia_address BYTEA NOT NULL CHECK (LENGTH(sia_address) = 32), + merkle_proof BYTEA NOT NULL, -- binary serialized []types.Hash256 + leaf_index BIGINT NOT NULL, + maturity_height BIGINT NOT NULL +); + +CREATE TABLE wallet_broadcasted_txnsets ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + basis BYTEA NOT NULL CHECK (LENGTH(basis) = 8+32), -- binary serialized chain index + raw_transactions BYTEA NOT NULL, -- binary serialized transaction set + date_created TIMESTAMP WITH TIME ZONE NOT NULL +); +CREATE INDEX wallet_broadcasted_txnsets_date_created ON wallet_broadcasted_txnsets(date_created); + +CREATE TABLE wallet_events ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + chain_index BYTEA NOT NULL CHECK (LENGTH(chain_index) = 8+32), + maturity_height BIGINT NOT NULL, + event_type TEXT NOT NULL, + raw_data BYTEA NOT NULL +); +CREATE INDEX wallet_events_chain_index ON wallet_events(chain_index); +CREATE INDEX wallet_events_maturity_height ON wallet_events(maturity_height DESC); + +CREATE TABLE stored_sectors ( + id BIGSERIAL PRIMARY KEY, + sector_root BYTEA UNIQUE NOT NULL CHECK (LENGTH(sector_root) = 32), + cached_subtree_roots BYTEA, + last_access_timestamp TIMESTAMP WITH TIME ZONE NOT NULL +); +CREATE INDEX stored_sectors_sector_root ON stored_sectors(sector_root); +CREATE INDEX stored_sectors_last_access ON stored_sectors(last_access_timestamp); + +CREATE TABLE storage_volumes ( + id BIGSERIAL PRIMARY KEY, + disk_path TEXT UNIQUE NOT NULL, + used_sectors BIGINT NOT NULL, + total_sectors BIGINT NOT NULL, + read_only BOOLEAN NOT NULL, + available BOOLEAN NOT NULL DEFAULT false +); +CREATE INDEX storage_volumes_id_available_read_only ON storage_volumes(id, available, read_only); +CREATE INDEX storage_volumes_read_only_available_used_sectors ON storage_volumes(available, read_only, used_sectors); + +CREATE TABLE volume_sectors ( + id BIGSERIAL PRIMARY KEY, + volume_id BIGINT NOT NULL REFERENCES storage_volumes (id), -- all sectors will need to be migrated first when deleting a volume + volume_index BIGINT NOT NULL, + sector_id BIGINT UNIQUE REFERENCES stored_sectors (id), + sector_writes BIGINT NOT NULL DEFAULT 0, + UNIQUE (volume_id, volume_index) +); +CREATE INDEX volume_sectors_sector_writes_volume_id_sector_id_volume_index_compound ON volume_sectors(sector_writes ASC, volume_id, sector_id, volume_index) WHERE sector_id IS NULL; +CREATE INDEX volume_sectors_volume_id_sector_id ON volume_sectors(volume_id, sector_id); +CREATE INDEX volume_sectors_volume_id ON volume_sectors(volume_id); +CREATE INDEX volume_sectors_volume_index ON volume_sectors(volume_index ASC); +CREATE INDEX volume_sectors_sector_id ON volume_sectors(sector_id); + +CREATE TABLE contract_renters ( + id BIGSERIAL PRIMARY KEY, + public_key BYTEA UNIQUE NOT NULL CHECK (LENGTH(public_key) = 32) +); + +CREATE TABLE contracts ( + id BIGSERIAL PRIMARY KEY, + renter_id BIGINT NOT NULL REFERENCES contract_renters(id), + renewed_to BIGINT REFERENCES contracts(id) ON DELETE SET NULL, + renewed_from BIGINT REFERENCES contracts(id) ON DELETE SET NULL, + contract_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(contract_id) = 32), + revision_number NUMERIC(20,0) NOT NULL, -- uint64, supports uint64_max on clearing revisions + formation_txn_set BYTEA NOT NULL, -- binary serialized transaction set + locked_collateral NUMERIC(50,0) NOT NULL, + rpc_revenue NUMERIC(50,0) NOT NULL, + storage_revenue NUMERIC(50,0) NOT NULL, + ingress_revenue NUMERIC(50,0) NOT NULL, + egress_revenue NUMERIC(50,0) NOT NULL, + account_funding NUMERIC(50,0) NOT NULL, + registry_read NUMERIC(50,0) NOT NULL, + registry_write NUMERIC(50,0) NOT NULL, + risked_collateral NUMERIC(50,0) NOT NULL, + confirmed_revision_number NUMERIC(20,0), -- uint64, supports uint64_max on clearing revisions + host_sig BYTEA NOT NULL CHECK (LENGTH(host_sig) = 64), + renter_sig BYTEA NOT NULL CHECK (LENGTH(renter_sig) = 64), + raw_revision BYTEA NOT NULL, -- binary serialized contract revision + formation_confirmed BOOLEAN NOT NULL, -- true if the contract has been confirmed on the blockchain + resolution_height BIGINT, -- null if the storage proof/resolution has not been confirmed on the blockchain, otherwise the height of the block containing the storage proof/resolution + negotiation_height BIGINT NOT NULL, -- determines if the formation txn should be rebroadcast or if the contract should be deleted + window_start BIGINT NOT NULL, + window_end BIGINT NOT NULL, + contract_status SMALLINT NOT NULL +); +CREATE INDEX contracts_contract_id ON contracts(contract_id); +CREATE INDEX contracts_renter_id ON contracts(renter_id); +CREATE INDEX contracts_renewed_to ON contracts(renewed_to); +CREATE INDEX contracts_renewed_from ON contracts(renewed_from); +CREATE INDEX contracts_negotiation_height ON contracts(negotiation_height); +CREATE INDEX contracts_window_start ON contracts(window_start); +CREATE INDEX contracts_window_end ON contracts(window_end); +CREATE INDEX contracts_contract_status ON contracts(contract_status); +CREATE INDEX contracts_formation_confirmed_resolution_height_window_start ON contracts(formation_confirmed, resolution_height, window_start); +CREATE INDEX contracts_formation_confirmed_resolution_height_window_end ON contracts(formation_confirmed, resolution_height, window_end); +CREATE INDEX contracts_formation_confirmed_window_start ON contracts(formation_confirmed, window_start); +CREATE INDEX contracts_formation_confirmed_negotiation_height ON contracts(formation_confirmed, negotiation_height); + +CREATE TABLE contract_sector_roots ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts(id), + sector_id BIGINT NOT NULL REFERENCES stored_sectors(id), + root_index BIGINT NOT NULL, + UNIQUE(contract_id, root_index) +); +CREATE INDEX contract_sector_roots_sector_id ON contract_sector_roots(sector_id); +CREATE INDEX contract_sector_roots_contract_id_root_index ON contract_sector_roots(contract_id, root_index); + +CREATE TABLE contract_v2_roots_map ( + id BIGINT NOT NULL, + revision_number BIGINT NOT NULL, + PRIMARY KEY (id, revision_number) +); + +CREATE TABLE contracts_v2 ( + id BIGSERIAL PRIMARY KEY, + renter_id BIGINT NOT NULL REFERENCES contract_renters(id), + renewed_to BIGINT REFERENCES contracts_v2(id) ON DELETE SET NULL, + renewed_from BIGINT REFERENCES contracts_v2(id) ON DELETE SET NULL, + contract_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(contract_id) = 32), + revision_number NUMERIC(20,0) NOT NULL, -- uint64, supports uint64_max on clearing revisions + formation_txn_set BYTEA NOT NULL, -- binary serialized transaction set + formation_txn_set_basis BYTEA NOT NULL CHECK (LENGTH(formation_txn_set_basis) = 8+32), + locked_collateral NUMERIC(50,0) NOT NULL, + rpc_revenue NUMERIC(50,0) NOT NULL, + storage_revenue NUMERIC(50,0) NOT NULL, + ingress_revenue NUMERIC(50,0) NOT NULL, + egress_revenue NUMERIC(50,0) NOT NULL, + account_funding NUMERIC(50,0) NOT NULL, + risked_collateral NUMERIC(50,0) NOT NULL, + raw_revision BYTEA NOT NULL, -- binary serialized contract revision + confirmation_index BYTEA CHECK (LENGTH(confirmation_index) = 8+32), -- null if the contract has not been confirmed on the blockchain, otherwise the chain index of the block containing the confirmation transaction + negotiation_height BIGINT NOT NULL, -- determines if the formation txn should be rebroadcast or if the contract should be deleted + proof_height BIGINT NOT NULL, + expiration_height BIGINT NOT NULL, + resolution_block_id BYTEA CHECK (LENGTH(resolution_block_id) = 32), -- null if the resolution has not been confirmed on the blockchain + resolution_height BIGINT CHECK((resolution_height IS NULL) = (resolution_block_id IS NULL)), -- null if the resolution has not been confirmed on the blockchain + contract_status TEXT NOT NULL, + sector_count BIGINT NOT NULL, -- used for cleanup + + contract_v2_roots_map_id BIGINT NOT NULL, + contract_v2_roots_map_revision_number BIGINT NOT NULL, + FOREIGN KEY (contract_v2_roots_map_id, contract_v2_roots_map_revision_number) REFERENCES contract_v2_roots_map(id, revision_number) +); +CREATE INDEX contracts_v2_contract_id ON contracts_v2(contract_id); +CREATE INDEX contracts_v2_renter_id ON contracts_v2(renter_id); +CREATE INDEX contracts_v2_renewed_to ON contracts_v2(renewed_to); +CREATE INDEX contracts_v2_renewed_from ON contracts_v2(renewed_from); +CREATE INDEX contracts_v2_negotiation_height ON contracts_v2(negotiation_height); +CREATE INDEX contracts_v2_proof_height ON contracts_v2(proof_height); +CREATE INDEX contracts_v2_expiration_height ON contracts_v2(expiration_height); +CREATE INDEX contracts_v2_contract_status ON contracts_v2(contract_status); +CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_proof_height ON contracts_v2(confirmation_index, resolution_block_id, proof_height); +CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_expiration_height ON contracts_v2(confirmation_index, resolution_block_id, expiration_height); +CREATE INDEX contracts_v2_resolution_height ON contracts_v2(resolution_height); +CREATE INDEX contracts_v2_confirmation_index_proof_height ON contracts_v2(confirmation_index, proof_height); +CREATE INDEX contracts_v2_confirmation_index_negotiation_height ON contracts_v2(confirmation_index, negotiation_height); +CREATE INDEX contracts_v2_roots_map_id_contract_v2_roots_map_revision_number ON contracts_v2(contract_v2_roots_map_id, contract_v2_roots_map_revision_number); + +CREATE TABLE contract_v2_state_elements ( + contract_id BIGINT PRIMARY KEY REFERENCES contracts_v2(id), + leaf_index BIGINT NOT NULL, + merkle_proof BYTEA NOT NULL, + raw_contract BYTEA NOT NULL, -- binary serialized contract + revision_number NUMERIC(20,0) NOT NULL -- for comparison +); + +CREATE TABLE contracts_v2_chain_index_elements ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + height BIGINT NOT NULL, + leaf_index BIGINT NOT NULL, + merkle_proof BYTEA NOT NULL +); +CREATE INDEX contracts_v2_chain_index_elements_height ON contracts_v2_chain_index_elements(height); + +CREATE TABLE contract_v2_sector_roots ( + id BIGSERIAL PRIMARY KEY, + sector_id BIGINT NOT NULL REFERENCES stored_sectors(id), + root_index BIGINT NOT NULL, + contract_v2_roots_map_id BIGINT NOT NULL, + contract_v2_roots_map_revision_number BIGINT NOT NULL, + FOREIGN KEY (contract_v2_roots_map_id, contract_v2_roots_map_revision_number) REFERENCES contract_v2_roots_map(id, revision_number), + UNIQUE(contract_v2_roots_map_id, contract_v2_roots_map_revision_number, root_index) +); +CREATE INDEX contract_v2_sector_roots_map_id_root_index_revision_number ON contract_v2_sector_roots(contract_v2_roots_map_id, root_index, contract_v2_roots_map_revision_number); +CREATE INDEX contract_v2_sector_roots_sector_id ON contract_v2_sector_roots(sector_id); + +CREATE TABLE temp_storage_sector_roots ( + id BIGSERIAL PRIMARY KEY, + sector_id BIGINT NOT NULL REFERENCES stored_sectors(id), + expiration_height BIGINT NOT NULL +); +CREATE INDEX temp_storage_sector_roots_sector_id ON temp_storage_sector_roots(sector_id); +CREATE INDEX temp_storage_sector_roots_expiration_height ON temp_storage_sector_roots(expiration_height); + +CREATE TABLE registry_entries ( + registry_key BYTEA PRIMARY KEY CHECK (LENGTH(registry_key) = 32), + revision_number NUMERIC(20,0) NOT NULL, -- uint64, supports uint64_max + entry_data BYTEA NOT NULL, + entry_signature BYTEA NOT NULL CHECK (LENGTH(entry_signature) = 64), + entry_type SMALLINT NOT NULL, + expiration_height BIGINT NOT NULL +); +CREATE INDEX registry_entries_expiration_height ON registry_entries(expiration_height); + +CREATE TABLE accounts ( + id BIGSERIAL PRIMARY KEY, + account_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(account_id) = 32), + balance NUMERIC(50,0) NOT NULL, + expiration_timestamp TIMESTAMP WITH TIME ZONE NOT NULL +); +CREATE INDEX accounts_expiration_timestamp ON accounts(expiration_timestamp); + +CREATE TABLE contract_account_funding ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts(id), + account_id BIGINT NOT NULL REFERENCES accounts(id), + amount NUMERIC(50,0) NOT NULL, + UNIQUE (contract_id, account_id) +); + +CREATE TABLE contract_v2_account_funding ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts_v2(id), + account_id BIGINT NOT NULL REFERENCES accounts(id), + amount NUMERIC(50,0) NOT NULL, + UNIQUE (contract_id, account_id) +); + +CREATE TABLE rhp4_pools ( + id BIGSERIAL PRIMARY KEY, + pool_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(pool_id) = 32), + balance NUMERIC(50,0) NOT NULL +); + +CREATE TABLE contract_v2_pool_funding ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts_v2(id), + pool_id BIGINT NOT NULL REFERENCES rhp4_pools(id), + amount NUMERIC(50,0) NOT NULL, + UNIQUE (contract_id, pool_id) +); +CREATE INDEX contract_v2_pool_funding_pool_id ON contract_v2_pool_funding(pool_id); + +CREATE TABLE rhp4_account_pool_attachments ( + id BIGSERIAL PRIMARY KEY, + account_id BIGINT NOT NULL REFERENCES accounts(id), + pool_id BIGINT NOT NULL REFERENCES rhp4_pools(id), + UNIQUE (account_id, pool_id) +); +CREATE INDEX rhp4_account_pool_attachments_account_id_id ON rhp4_account_pool_attachments(account_id, id); +CREATE INDEX rhp4_account_pool_attachments_pool_id ON rhp4_account_pool_attachments(pool_id); + +CREATE TABLE host_stats ( + date_created TIMESTAMP WITH TIME ZONE NOT NULL, + stat TEXT NOT NULL, + stat_value BYTEA NOT NULL, -- polymorphic: currency, uint64, or float64 bits depending on stat + PRIMARY KEY(date_created, stat) +); +CREATE INDEX host_stats_stat_date_created ON host_stats(stat, date_created DESC); + +CREATE TABLE host_settings ( + id BIGINT PRIMARY KEY NOT NULL DEFAULT 0 CHECK (id = 0), -- enforce a single row + settings_revision BIGINT NOT NULL, + accepting_contracts BOOLEAN NOT NULL, + net_address TEXT NOT NULL, + contract_price NUMERIC(50,0) NOT NULL, + base_rpc_price NUMERIC(50,0) NOT NULL, + sector_access_price NUMERIC(50,0) NOT NULL, + max_collateral NUMERIC(50,0) NOT NULL, + storage_price NUMERIC(50,0) NOT NULL, + egress_price NUMERIC(50,0) NOT NULL, + ingress_price NUMERIC(50,0) NOT NULL, + max_account_balance NUMERIC(50,0) NOT NULL, + collateral_multiplier DOUBLE PRECISION NOT NULL, + max_account_age BIGINT NOT NULL, -- time.Duration + price_table_validity BIGINT NOT NULL, -- time.Duration + max_contract_duration BIGINT NOT NULL, + window_size BIGINT NOT NULL, + ingress_limit BIGINT NOT NULL, + egress_limit BIGINT NOT NULL, + syncer_ingress_limit BIGINT NOT NULL, + syncer_egress_limit BIGINT NOT NULL, + ddns_provider TEXT NOT NULL, + ddns_update_v4 BOOLEAN NOT NULL, + ddns_update_v6 BOOLEAN NOT NULL, + ddns_opts BYTEA, + registry_limit BIGINT NOT NULL, + sector_cache_size BIGINT NOT NULL DEFAULT 0 +); + +CREATE TABLE host_pinned_settings ( + id BIGINT PRIMARY KEY NOT NULL DEFAULT 0 CHECK (id = 0), -- enforce a single row + currency TEXT NOT NULL, + threshold DOUBLE PRECISION NOT NULL, + storage_pinned BOOLEAN NOT NULL, + storage_price DOUBLE PRECISION NOT NULL, + ingress_pinned BOOLEAN NOT NULL, + ingress_price DOUBLE PRECISION NOT NULL, + egress_pinned BOOLEAN NOT NULL, + egress_price DOUBLE PRECISION NOT NULL, + max_collateral_pinned BOOLEAN NOT NULL, + max_collateral DOUBLE PRECISION NOT NULL +); + +CREATE TABLE webhooks ( + id BIGSERIAL PRIMARY KEY, + callback_url TEXT UNIQUE NOT NULL, + scopes TEXT NOT NULL, + secret_key TEXT UNIQUE NOT NULL +); + +CREATE TABLE syncer_peers ( + peer_address TEXT PRIMARY KEY NOT NULL, + first_seen TIMESTAMP WITH TIME ZONE NOT NULL +); + +CREATE TABLE syncer_bans ( + net_cidr TEXT PRIMARY KEY NOT NULL, + expiration TIMESTAMP WITH TIME ZONE NOT NULL, + reason TEXT NOT NULL +); +CREATE INDEX syncer_bans_expiration_index_idx ON syncer_bans (expiration); + +CREATE TABLE global_settings ( + id BIGINT PRIMARY KEY NOT NULL DEFAULT 0 CHECK (id = 0), -- enforce a single row + db_version BIGINT NOT NULL, -- used for migrations + host_key BYTEA CHECK (LENGTH(host_key) = 64), -- ed25519 private key + wallet_hash BYTEA CHECK (LENGTH(wallet_hash) = 32), -- used to prevent wallet seed changes + last_scanned_index BYTEA CHECK (LENGTH(last_scanned_index) = 8+32), -- chain index of the last scanned block + last_announce_index BYTEA CHECK (LENGTH(last_announce_index) = 8+32), -- chain index of the last host announcement + last_announce_address TEXT, -- address of the last host announcement + last_v2_announce_hash BYTEA CHECK (LENGTH(last_v2_announce_hash) = 32) -- hash of the last v2 host announcement +); + +-- initialize the global settings table +INSERT INTO global_settings (id, db_version) VALUES (0, 0); -- should not be changed diff --git a/persist/postgres/migrations.go b/persist/postgres/migrations.go new file mode 100644 index 00000000..976a0bfe --- /dev/null +++ b/persist/postgres/migrations.go @@ -0,0 +1,15 @@ +package postgres + +import ( + "context" + + "go.uber.org/zap" +) + +// migrations is a list of functions that are run to migrate the database from +// one version to the next. Migrations are used to update existing databases to +// match the schema in init.sql. +// +// The PostgreSQL store starts at version 1 (the initial schema) and has no +// migrations yet. +var migrations = []func(ctx context.Context, tx *txn, log *zap.Logger) error{} diff --git a/persist/postgres/migrations_test.go b/persist/postgres/migrations_test.go new file mode 100644 index 00000000..9551867b --- /dev/null +++ b/persist/postgres/migrations_test.go @@ -0,0 +1,743 @@ +package postgres + +import ( + "context" + "encoding/hex" + "fmt" + "io" + "net" + "os" + "path/filepath" + "testing" + + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + "go.uber.org/zap/zaptest" + "lukechampine.com/frand" +) + +// initialSchema is a frozen copy of the version 1 schema (init.sql at the time +// the first migration was introduced). It is used by TestMigrationConsistency to +// build a database at the oldest supported version so that the migration path up +// to the current schema can be exercised and compared against a database +// initialized directly from init.sql. +// +// IMPORTANT: this must NOT be updated when init.sql changes. Schema changes are +// applied via migrations in migrations.go; this constant preserves the starting +// point those migrations run against. +const initialSchema = ` +CREATE TABLE wallet_siacoin_elements ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + siacoin_value NUMERIC(50,0) NOT NULL, + sia_address BYTEA NOT NULL CHECK (LENGTH(sia_address) = 32), + merkle_proof BYTEA NOT NULL, -- binary serialized []types.Hash256 + leaf_index BIGINT NOT NULL, + maturity_height BIGINT NOT NULL +); + +CREATE TABLE wallet_broadcasted_txnsets ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + basis BYTEA NOT NULL CHECK (LENGTH(basis) = 8+32), -- binary serialized chain index + raw_transactions BYTEA NOT NULL, -- binary serialized transaction set + date_created TIMESTAMP WITH TIME ZONE NOT NULL +); +CREATE INDEX wallet_broadcasted_txnsets_date_created ON wallet_broadcasted_txnsets(date_created); + +CREATE TABLE wallet_events ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + chain_index BYTEA NOT NULL CHECK (LENGTH(chain_index) = 8+32), + maturity_height BIGINT NOT NULL, + event_type TEXT NOT NULL, + raw_data BYTEA NOT NULL +); +CREATE INDEX wallet_events_chain_index ON wallet_events(chain_index); +CREATE INDEX wallet_events_maturity_height ON wallet_events(maturity_height DESC); + +CREATE TABLE stored_sectors ( + id BIGSERIAL PRIMARY KEY, + sector_root BYTEA UNIQUE NOT NULL CHECK (LENGTH(sector_root) = 32), + cached_subtree_roots BYTEA, + last_access_timestamp TIMESTAMP WITH TIME ZONE NOT NULL +); +CREATE INDEX stored_sectors_sector_root ON stored_sectors(sector_root); +CREATE INDEX stored_sectors_last_access ON stored_sectors(last_access_timestamp); + +CREATE TABLE storage_volumes ( + id BIGSERIAL PRIMARY KEY, + disk_path TEXT UNIQUE NOT NULL, + used_sectors BIGINT NOT NULL, + total_sectors BIGINT NOT NULL, + read_only BOOLEAN NOT NULL, + available BOOLEAN NOT NULL DEFAULT false +); +CREATE INDEX storage_volumes_id_available_read_only ON storage_volumes(id, available, read_only); +CREATE INDEX storage_volumes_read_only_available_used_sectors ON storage_volumes(available, read_only, used_sectors); + +CREATE TABLE volume_sectors ( + id BIGSERIAL PRIMARY KEY, + volume_id BIGINT NOT NULL REFERENCES storage_volumes (id), -- all sectors will need to be migrated first when deleting a volume + volume_index BIGINT NOT NULL, + sector_id BIGINT UNIQUE REFERENCES stored_sectors (id), + sector_writes BIGINT NOT NULL DEFAULT 0, + UNIQUE (volume_id, volume_index) +); +CREATE INDEX volume_sectors_sector_writes_volume_id_sector_id_volume_index_compound ON volume_sectors(sector_writes ASC, volume_id, sector_id, volume_index) WHERE sector_id IS NULL; +CREATE INDEX volume_sectors_volume_id_sector_id ON volume_sectors(volume_id, sector_id); +CREATE INDEX volume_sectors_volume_id ON volume_sectors(volume_id); +CREATE INDEX volume_sectors_volume_index ON volume_sectors(volume_index ASC); +CREATE INDEX volume_sectors_sector_id ON volume_sectors(sector_id); + +CREATE TABLE contract_renters ( + id BIGSERIAL PRIMARY KEY, + public_key BYTEA UNIQUE NOT NULL CHECK (LENGTH(public_key) = 32) +); + +CREATE TABLE contracts ( + id BIGSERIAL PRIMARY KEY, + renter_id BIGINT NOT NULL REFERENCES contract_renters(id), + renewed_to BIGINT REFERENCES contracts(id) ON DELETE SET NULL, + renewed_from BIGINT REFERENCES contracts(id) ON DELETE SET NULL, + contract_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(contract_id) = 32), + revision_number NUMERIC(20,0) NOT NULL, -- uint64, supports uint64_max on clearing revisions + formation_txn_set BYTEA NOT NULL, -- binary serialized transaction set + locked_collateral NUMERIC(50,0) NOT NULL, + rpc_revenue NUMERIC(50,0) NOT NULL, + storage_revenue NUMERIC(50,0) NOT NULL, + ingress_revenue NUMERIC(50,0) NOT NULL, + egress_revenue NUMERIC(50,0) NOT NULL, + account_funding NUMERIC(50,0) NOT NULL, + registry_read NUMERIC(50,0) NOT NULL, + registry_write NUMERIC(50,0) NOT NULL, + risked_collateral NUMERIC(50,0) NOT NULL, + confirmed_revision_number NUMERIC(20,0), -- uint64, supports uint64_max on clearing revisions + host_sig BYTEA NOT NULL CHECK (LENGTH(host_sig) = 64), + renter_sig BYTEA NOT NULL CHECK (LENGTH(renter_sig) = 64), + raw_revision BYTEA NOT NULL, -- binary serialized contract revision + formation_confirmed BOOLEAN NOT NULL, -- true if the contract has been confirmed on the blockchain + resolution_height BIGINT, -- null if the storage proof/resolution has not been confirmed on the blockchain, otherwise the height of the block containing the storage proof/resolution + negotiation_height BIGINT NOT NULL, -- determines if the formation txn should be rebroadcast or if the contract should be deleted + window_start BIGINT NOT NULL, + window_end BIGINT NOT NULL, + contract_status SMALLINT NOT NULL +); +CREATE INDEX contracts_contract_id ON contracts(contract_id); +CREATE INDEX contracts_renter_id ON contracts(renter_id); +CREATE INDEX contracts_renewed_to ON contracts(renewed_to); +CREATE INDEX contracts_renewed_from ON contracts(renewed_from); +CREATE INDEX contracts_negotiation_height ON contracts(negotiation_height); +CREATE INDEX contracts_window_start ON contracts(window_start); +CREATE INDEX contracts_window_end ON contracts(window_end); +CREATE INDEX contracts_contract_status ON contracts(contract_status); +CREATE INDEX contracts_formation_confirmed_resolution_height_window_start ON contracts(formation_confirmed, resolution_height, window_start); +CREATE INDEX contracts_formation_confirmed_resolution_height_window_end ON contracts(formation_confirmed, resolution_height, window_end); +CREATE INDEX contracts_formation_confirmed_window_start ON contracts(formation_confirmed, window_start); +CREATE INDEX contracts_formation_confirmed_negotiation_height ON contracts(formation_confirmed, negotiation_height); + +CREATE TABLE contract_sector_roots ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts(id), + sector_id BIGINT NOT NULL REFERENCES stored_sectors(id), + root_index BIGINT NOT NULL, + UNIQUE(contract_id, root_index) +); +CREATE INDEX contract_sector_roots_sector_id ON contract_sector_roots(sector_id); +CREATE INDEX contract_sector_roots_contract_id_root_index ON contract_sector_roots(contract_id, root_index); + +CREATE TABLE contract_v2_roots_map ( + id BIGINT NOT NULL, + revision_number BIGINT NOT NULL, + PRIMARY KEY (id, revision_number) +); + +CREATE TABLE contracts_v2 ( + id BIGSERIAL PRIMARY KEY, + renter_id BIGINT NOT NULL REFERENCES contract_renters(id), + renewed_to BIGINT REFERENCES contracts_v2(id) ON DELETE SET NULL, + renewed_from BIGINT REFERENCES contracts_v2(id) ON DELETE SET NULL, + contract_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(contract_id) = 32), + revision_number NUMERIC(20,0) NOT NULL, -- uint64, supports uint64_max on clearing revisions + formation_txn_set BYTEA NOT NULL, -- binary serialized transaction set + formation_txn_set_basis BYTEA NOT NULL CHECK (LENGTH(formation_txn_set_basis) = 8+32), + locked_collateral NUMERIC(50,0) NOT NULL, + rpc_revenue NUMERIC(50,0) NOT NULL, + storage_revenue NUMERIC(50,0) NOT NULL, + ingress_revenue NUMERIC(50,0) NOT NULL, + egress_revenue NUMERIC(50,0) NOT NULL, + account_funding NUMERIC(50,0) NOT NULL, + risked_collateral NUMERIC(50,0) NOT NULL, + raw_revision BYTEA NOT NULL, -- binary serialized contract revision + confirmation_index BYTEA CHECK (LENGTH(confirmation_index) = 8+32), -- null if the contract has not been confirmed on the blockchain, otherwise the chain index of the block containing the confirmation transaction + negotiation_height BIGINT NOT NULL, -- determines if the formation txn should be rebroadcast or if the contract should be deleted + proof_height BIGINT NOT NULL, + expiration_height BIGINT NOT NULL, + resolution_block_id BYTEA CHECK (LENGTH(resolution_block_id) = 32), -- null if the resolution has not been confirmed on the blockchain + resolution_height BIGINT CHECK((resolution_height IS NULL) = (resolution_block_id IS NULL)), -- null if the resolution has not been confirmed on the blockchain + contract_status TEXT NOT NULL, + sector_count BIGINT NOT NULL, -- used for cleanup + + contract_v2_roots_map_id BIGINT NOT NULL, + contract_v2_roots_map_revision_number BIGINT NOT NULL, + FOREIGN KEY (contract_v2_roots_map_id, contract_v2_roots_map_revision_number) REFERENCES contract_v2_roots_map(id, revision_number) +); +CREATE INDEX contracts_v2_contract_id ON contracts_v2(contract_id); +CREATE INDEX contracts_v2_renter_id ON contracts_v2(renter_id); +CREATE INDEX contracts_v2_renewed_to ON contracts_v2(renewed_to); +CREATE INDEX contracts_v2_renewed_from ON contracts_v2(renewed_from); +CREATE INDEX contracts_v2_negotiation_height ON contracts_v2(negotiation_height); +CREATE INDEX contracts_v2_proof_height ON contracts_v2(proof_height); +CREATE INDEX contracts_v2_expiration_height ON contracts_v2(expiration_height); +CREATE INDEX contracts_v2_contract_status ON contracts_v2(contract_status); +CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_proof_height ON contracts_v2(confirmation_index, resolution_block_id, proof_height); +CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_expiration_height ON contracts_v2(confirmation_index, resolution_block_id, expiration_height); +CREATE INDEX contracts_v2_resolution_height ON contracts_v2(resolution_height); +CREATE INDEX contracts_v2_confirmation_index_proof_height ON contracts_v2(confirmation_index, proof_height); +CREATE INDEX contracts_v2_confirmation_index_negotiation_height ON contracts_v2(confirmation_index, negotiation_height); +CREATE INDEX contracts_v2_roots_map_id_contract_v2_roots_map_revision_number ON contracts_v2(contract_v2_roots_map_id, contract_v2_roots_map_revision_number); + +CREATE TABLE contract_v2_state_elements ( + contract_id BIGINT PRIMARY KEY REFERENCES contracts_v2(id), + leaf_index BIGINT NOT NULL, + merkle_proof BYTEA NOT NULL, + raw_contract BYTEA NOT NULL, -- binary serialized contract + revision_number NUMERIC(20,0) NOT NULL -- for comparison +); + +CREATE TABLE contracts_v2_chain_index_elements ( + id BYTEA PRIMARY KEY CHECK (LENGTH(id) = 32), + height BIGINT NOT NULL, + leaf_index BIGINT NOT NULL, + merkle_proof BYTEA NOT NULL +); +CREATE INDEX contracts_v2_chain_index_elements_height ON contracts_v2_chain_index_elements(height); + +CREATE TABLE contract_v2_sector_roots ( + id BIGSERIAL PRIMARY KEY, + sector_id BIGINT NOT NULL REFERENCES stored_sectors(id), + root_index BIGINT NOT NULL, + contract_v2_roots_map_id BIGINT NOT NULL, + contract_v2_roots_map_revision_number BIGINT NOT NULL, + FOREIGN KEY (contract_v2_roots_map_id, contract_v2_roots_map_revision_number) REFERENCES contract_v2_roots_map(id, revision_number), + UNIQUE(contract_v2_roots_map_id, contract_v2_roots_map_revision_number, root_index) +); +CREATE INDEX contract_v2_sector_roots_map_id_root_index_revision_number ON contract_v2_sector_roots(contract_v2_roots_map_id, root_index, contract_v2_roots_map_revision_number); +CREATE INDEX contract_v2_sector_roots_sector_id ON contract_v2_sector_roots(sector_id); + +CREATE TABLE temp_storage_sector_roots ( + id BIGSERIAL PRIMARY KEY, + sector_id BIGINT NOT NULL REFERENCES stored_sectors(id), + expiration_height BIGINT NOT NULL +); +CREATE INDEX temp_storage_sector_roots_sector_id ON temp_storage_sector_roots(sector_id); +CREATE INDEX temp_storage_sector_roots_expiration_height ON temp_storage_sector_roots(expiration_height); + +CREATE TABLE registry_entries ( + registry_key BYTEA PRIMARY KEY CHECK (LENGTH(registry_key) = 32), + revision_number NUMERIC(20,0) NOT NULL, -- uint64, supports uint64_max + entry_data BYTEA NOT NULL, + entry_signature BYTEA NOT NULL CHECK (LENGTH(entry_signature) = 64), + entry_type SMALLINT NOT NULL, + expiration_height BIGINT NOT NULL +); +CREATE INDEX registry_entries_expiration_height ON registry_entries(expiration_height); + +CREATE TABLE accounts ( + id BIGSERIAL PRIMARY KEY, + account_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(account_id) = 32), + balance NUMERIC(50,0) NOT NULL, + expiration_timestamp TIMESTAMP WITH TIME ZONE NOT NULL +); +CREATE INDEX accounts_expiration_timestamp ON accounts(expiration_timestamp); + +CREATE TABLE contract_account_funding ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts(id), + account_id BIGINT NOT NULL REFERENCES accounts(id), + amount NUMERIC(50,0) NOT NULL, + UNIQUE (contract_id, account_id) +); + +CREATE TABLE contract_v2_account_funding ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts_v2(id), + account_id BIGINT NOT NULL REFERENCES accounts(id), + amount NUMERIC(50,0) NOT NULL, + UNIQUE (contract_id, account_id) +); + +CREATE TABLE rhp4_pools ( + id BIGSERIAL PRIMARY KEY, + pool_id BYTEA UNIQUE NOT NULL CHECK (LENGTH(pool_id) = 32), + balance NUMERIC(50,0) NOT NULL +); + +CREATE TABLE contract_v2_pool_funding ( + id BIGSERIAL PRIMARY KEY, + contract_id BIGINT NOT NULL REFERENCES contracts_v2(id), + pool_id BIGINT NOT NULL REFERENCES rhp4_pools(id), + amount NUMERIC(50,0) NOT NULL, + UNIQUE (contract_id, pool_id) +); +CREATE INDEX contract_v2_pool_funding_pool_id ON contract_v2_pool_funding(pool_id); + +CREATE TABLE rhp4_account_pool_attachments ( + id BIGSERIAL PRIMARY KEY, + account_id BIGINT NOT NULL REFERENCES accounts(id), + pool_id BIGINT NOT NULL REFERENCES rhp4_pools(id), + UNIQUE (account_id, pool_id) +); +CREATE INDEX rhp4_account_pool_attachments_account_id_id ON rhp4_account_pool_attachments(account_id, id); +CREATE INDEX rhp4_account_pool_attachments_pool_id ON rhp4_account_pool_attachments(pool_id); + +CREATE TABLE host_stats ( + date_created TIMESTAMP WITH TIME ZONE NOT NULL, + stat TEXT NOT NULL, + stat_value BYTEA NOT NULL, -- polymorphic: currency, uint64, or float64 bits depending on stat + PRIMARY KEY(date_created, stat) +); +CREATE INDEX host_stats_stat_date_created ON host_stats(stat, date_created DESC); + +CREATE TABLE host_settings ( + id BIGINT PRIMARY KEY NOT NULL DEFAULT 0 CHECK (id = 0), -- enforce a single row + settings_revision BIGINT NOT NULL, + accepting_contracts BOOLEAN NOT NULL, + net_address TEXT NOT NULL, + contract_price NUMERIC(50,0) NOT NULL, + base_rpc_price NUMERIC(50,0) NOT NULL, + sector_access_price NUMERIC(50,0) NOT NULL, + max_collateral NUMERIC(50,0) NOT NULL, + storage_price NUMERIC(50,0) NOT NULL, + egress_price NUMERIC(50,0) NOT NULL, + ingress_price NUMERIC(50,0) NOT NULL, + max_account_balance NUMERIC(50,0) NOT NULL, + collateral_multiplier DOUBLE PRECISION NOT NULL, + max_account_age BIGINT NOT NULL, -- time.Duration + price_table_validity BIGINT NOT NULL, -- time.Duration + max_contract_duration BIGINT NOT NULL, + window_size BIGINT NOT NULL, + ingress_limit BIGINT NOT NULL, + egress_limit BIGINT NOT NULL, + syncer_ingress_limit BIGINT NOT NULL, + syncer_egress_limit BIGINT NOT NULL, + ddns_provider TEXT NOT NULL, + ddns_update_v4 BOOLEAN NOT NULL, + ddns_update_v6 BOOLEAN NOT NULL, + ddns_opts BYTEA, + registry_limit BIGINT NOT NULL, + sector_cache_size BIGINT NOT NULL DEFAULT 0 +); + +CREATE TABLE host_pinned_settings ( + id BIGINT PRIMARY KEY NOT NULL DEFAULT 0 CHECK (id = 0), -- enforce a single row + currency TEXT NOT NULL, + threshold DOUBLE PRECISION NOT NULL, + storage_pinned BOOLEAN NOT NULL, + storage_price DOUBLE PRECISION NOT NULL, + ingress_pinned BOOLEAN NOT NULL, + ingress_price DOUBLE PRECISION NOT NULL, + egress_pinned BOOLEAN NOT NULL, + egress_price DOUBLE PRECISION NOT NULL, + max_collateral_pinned BOOLEAN NOT NULL, + max_collateral DOUBLE PRECISION NOT NULL +); + +CREATE TABLE webhooks ( + id BIGSERIAL PRIMARY KEY, + callback_url TEXT UNIQUE NOT NULL, + scopes TEXT NOT NULL, + secret_key TEXT UNIQUE NOT NULL +); + +CREATE TABLE syncer_peers ( + peer_address TEXT PRIMARY KEY NOT NULL, + first_seen TIMESTAMP WITH TIME ZONE NOT NULL +); + +CREATE TABLE syncer_bans ( + net_cidr TEXT PRIMARY KEY NOT NULL, + expiration TIMESTAMP WITH TIME ZONE NOT NULL, + reason TEXT NOT NULL +); +CREATE INDEX syncer_bans_expiration_index_idx ON syncer_bans (expiration); + +CREATE TABLE global_settings ( + id BIGINT PRIMARY KEY NOT NULL DEFAULT 0 CHECK (id = 0), -- enforce a single row + db_version BIGINT NOT NULL, -- used for migrations + host_key BYTEA CHECK (LENGTH(host_key) = 64), -- ed25519 private key + wallet_hash BYTEA CHECK (LENGTH(wallet_hash) = 32), -- used to prevent wallet seed changes + last_scanned_index BYTEA CHECK (LENGTH(last_scanned_index) = 8+32), -- chain index of the last scanned block + last_announce_index BYTEA CHECK (LENGTH(last_announce_index) = 8+32), -- chain index of the last host announcement + last_announce_address TEXT, -- address of the last host announcement + last_v2_announce_hash BYTEA CHECK (LENGTH(last_v2_announce_hash) = 32) -- hash of the last v2 host announcement +); +` + +// testConnInfo is the connection info for the embedded PostgreSQL server started +// in TestMain. Each test creates its own uniquely-named database. +var testConnInfo ConnectionInfo + +func TestMain(m *testing.M) { + os.Exit(runMain(m)) +} + +// runMain starts an embedded PostgreSQL server, runs the tests and shuts the +// server down. It is a separate function so that deferred cleanup runs before +// os.Exit. +func runMain(m *testing.M) int { + port, err := freePort() + if err != nil { + fmt.Fprintf(os.Stderr, "failed to find free port: %v\n", err) + return 1 + } + + runtimePath, err := os.MkdirTemp("", "hostd-postgres-*") + if err != nil { + fmt.Fprintf(os.Stderr, "failed to create runtime dir: %v\n", err) + return 1 + } + defer os.RemoveAll(runtimePath) + + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Username("postgres"). + Password("postgres"). + Database("postgres"). + Port(uint32(port)). + RuntimePath(runtimePath). + DataPath(filepath.Join(runtimePath, "data")). + Logger(io.Discard)) + if err := pg.Start(); err != nil { + fmt.Fprintf(os.Stderr, "failed to start embedded postgres: %v\n", err) + return 1 + } + defer func() { + if err := pg.Stop(); err != nil { + fmt.Fprintf(os.Stderr, "failed to stop embedded postgres: %v\n", err) + } + }() + + testConnInfo = ConnectionInfo{ + Host: "localhost", + Port: port, + User: "postgres", + Password: "postgres", + Database: "postgres", + SSLMode: "disable", + } + return m.Run() +} + +// freePort returns a free TCP port on the loopback interface. +func freePort() (int, error) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return 0, err + } + defer l.Close() + return l.Addr().(*net.TCPAddr).Port, nil +} + +// newConnInfo returns connection info for a fresh, uniquely-named database on +// the embedded server. +func newConnInfo() ConnectionInfo { + ci := testConnInfo + ci.Database = "hostd_" + hex.EncodeToString(frand.Bytes(8)) + return ci +} + +// openTestStore opens a store backed by a fresh database on the embedded server +// and registers cleanup to drop it. +func openTestStore(t *testing.T, log *zap.Logger) *Store { + t.Helper() + ci := newConnInfo() + store, err := OpenDatabase(context.Background(), ci, log) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + store.Close() + dropDatabase(t, ci) + }) + return store +} + +// dropDatabase drops the database referenced by ci by connecting to the default +// postgres database. +func dropDatabase(t *testing.T, ci ConnectionInfo) { + t.Helper() + admin := ci + admin.Database = "postgres" + pool, err := pgxpool.New(context.Background(), admin.connString()) + if err != nil { + t.Fatalf("failed to open admin connection: %v", err) + } + defer pool.Close() + if _, err := pool.Exec(context.Background(), fmt.Sprintf("DROP DATABASE IF EXISTS %s WITH (FORCE)", pgx.Identifier{ci.Database}.Sanitize())); err != nil { + t.Fatalf("failed to drop database %q: %v", ci.Database, err) + } +} + +func TestMigrationConsistency(t *testing.T) { + ctx := context.Background() + log := zaptest.NewLogger(t) + + ci := newConnInfo() + if err := ensureDatabase(ctx, ci); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { dropDatabase(t, ci) }) + + setupPool, err := pgxpool.New(ctx, ci.connString()) + if err != nil { + t.Fatal(err) + } + if _, err := setupPool.Exec(ctx, initialSchema); err != nil { + setupPool.Close() + t.Fatalf("failed to apply initial schema: %v", err) + } else if _, err := setupPool.Exec(ctx, `INSERT INTO global_settings (id, db_version) VALUES (0, 1)`); err != nil { + setupPool.Close() + t.Fatalf("failed to set initial version: %v", err) + } + setupPool.Close() + + // open the store, migrating the initial schema up to the target version + store, err := OpenDatabase(ctx, ci, log) + if err != nil { + t.Fatal(err) + } + + expectedVersion := int64(len(migrations) + 1) + if v := getDBVersion(ctx, store.pool); v != expectedVersion { + store.Close() + t.Fatalf("expected version %d, got %d", expectedVersion, v) + } + + // ensure the database does not change version when opened again + if err := store.Close(); err != nil { + t.Fatal(err) + } + store, err = OpenDatabase(ctx, ci, log) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if v := getDBVersion(ctx, store.pool); v != expectedVersion { + t.Fatalf("expected version %d after reopen, got %d", expectedVersion, v) + } + + // prepare a baseline database initialized directly from init.sql + baseline := openTestStore(t, log) + + // ensure the migrated database has the same indices as the baseline + baselineIndices, err := getTableIndices(ctx, baseline.pool) + if err != nil { + t.Fatal(err) + } + migratedIndices, err := getTableIndices(ctx, store.pool) + if err != nil { + t.Fatal(err) + } + for k := range baselineIndices { + if !migratedIndices[k] { + t.Errorf("missing index %s", k) + } + } + for k := range migratedIndices { + if !baselineIndices[k] { + t.Errorf("unexpected index %s", k) + } + } + + // ensure the migrated database has the same tables as the baseline + baselineTables, err := getTables(ctx, baseline.pool) + if err != nil { + t.Fatal(err) + } + migratedTables, err := getTables(ctx, store.pool) + if err != nil { + t.Fatal(err) + } + for k := range baselineTables { + if !migratedTables[k] { + t.Errorf("missing table %s", k) + } + } + for k := range migratedTables { + if !baselineTables[k] { + t.Errorf("unexpected table %s", k) + } + } + + // ensure each table has the same columns as the baseline + for k := range baselineTables { + baselineColumns, err := getTableColumns(ctx, baseline.pool, k) + if err != nil { + t.Fatal(err) + } + migratedColumns, err := getTableColumns(ctx, store.pool, k) + if err != nil { + t.Fatal(err) + } + for c := range baselineColumns { + if !migratedColumns[c] { + t.Errorf("missing column %s.%s", k, c) + } + } + for c := range migratedColumns { + if !baselineColumns[c] { + t.Errorf("unexpected column %s.%s", k, c) + } + } + } + + // ensure the migrated database has the same key constraints as the baseline + baselineKeyConstraints, err := getKeyConstraints(ctx, baseline.pool) + if err != nil { + t.Fatal(err) + } + migratedKeyConstraints, err := getKeyConstraints(ctx, store.pool) + if err != nil { + t.Fatal(err) + } + for kc := range baselineKeyConstraints { + if !migratedKeyConstraints[kc] { + t.Errorf("missing key constraint %s", kc) + } + } + for kc := range migratedKeyConstraints { + if !baselineKeyConstraints[kc] { + t.Errorf("unexpected key constraint %s", kc) + } + } + + // ensure the migrated database has the same check constraints as the baseline + baselineCheckConstraints, err := getCheckConstraints(ctx, baseline.pool) + if err != nil { + t.Fatal(err) + } + migratedCheckConstraints, err := getCheckConstraints(ctx, store.pool) + if err != nil { + t.Fatal(err) + } + for cc := range baselineCheckConstraints { + if !migratedCheckConstraints[cc] { + t.Errorf("missing check constraint %s", cc) + } + } + for cc := range migratedCheckConstraints { + if !baselineCheckConstraints[cc] { + t.Errorf("unexpected check constraint %s", cc) + } + } +} + +func getTableIndices(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) { + // https://www.postgresql.org/docs/current/view-pg-indexes.html + const query = `SELECT schemaname, tablename, indexname, tablespace, indexdef FROM pg_indexes WHERE schemaname = 'public'` + rows, err := pool.Query(ctx, query) + if err != nil { + return nil, err + } + defer rows.Close() + + indices := make(map[string]bool) + for rows.Next() { + var schema, table, index, def string + var tablespace *string // tablespace is null if default for the database + if err := rows.Scan(&schema, &table, &index, &tablespace, &def); err != nil { + return nil, err + } + var ts string + if tablespace != nil { + ts = *tablespace + } + indices[fmt.Sprintf("%s.%s.%s.%s.%s", schema, table, index, ts, def)] = true + } + return indices, rows.Err() +} + +func getTables(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) { + // https://www.postgresql.org/docs/current/infoschema-tables.html + const query = `SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'` + rows, err := pool.Query(ctx, query) + if err != nil { + return nil, err + } + defer rows.Close() + + tables := make(map[string]bool) + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, err + } + tables[name] = true + } + return tables, rows.Err() +} + +func getTableColumns(ctx context.Context, pool *pgxpool.Pool, table string) (map[string]bool, error) { + // https://www.postgresql.org/docs/current/infoschema-columns.html + const query = `SELECT column_name, data_type, column_default, is_nullable FROM information_schema.columns WHERE table_schema = 'public' AND table_name = $1` + rows, err := pool.Query(ctx, query, table) + if err != nil { + return nil, err + } + defer rows.Close() + + columns := make(map[string]bool) + for rows.Next() { + var name, colType, nullable string + var colDefault *string + if err := rows.Scan(&name, &colType, &colDefault, &nullable); err != nil { + return nil, err + } + var def string + if colDefault != nil { + def = *colDefault + } + columns[fmt.Sprintf("%s.%s.%s.%s", name, colType, def, nullable)] = true + } + return columns, rows.Err() +} + +func getKeyConstraints(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) { + // https://www.postgresql.org/docs/current/infoschema-key-column-usage.html + const query = `SELECT constraint_schema, constraint_name, table_name, column_name FROM information_schema.key_column_usage WHERE constraint_schema = 'public'` + rows, err := pool.Query(ctx, query) + if err != nil { + return nil, err + } + defer rows.Close() + + constraints := make(map[string]bool) + for rows.Next() { + var schema, name, table, column string + if err := rows.Scan(&schema, &name, &table, &column); err != nil { + return nil, err + } + constraints[fmt.Sprintf("%s.%s.%s.%s", schema, name, table, column)] = true + } + return constraints, rows.Err() +} + +func getCheckConstraints(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) { + // https://www.postgresql.org/docs/current/infoschema-check-constraints.html + const query = `SELECT constraint_schema, check_clause FROM information_schema.check_constraints WHERE constraint_schema = 'public'` + rows, err := pool.Query(ctx, query) + if err != nil { + return nil, err + } + defer rows.Close() + + constraints := make(map[string]bool) + for rows.Next() { + var schema, clause string + if err := rows.Scan(&schema, &clause); err != nil { + return nil, err + } + // the constraint name is ignored since it doesn't match between databases + constraints[fmt.Sprintf("%s.%s", schema, clause)] = true + } + return constraints, rows.Err() +} diff --git a/persist/postgres/sql.go b/persist/postgres/sql.go new file mode 100644 index 00000000..eac9101c --- /dev/null +++ b/persist/postgres/sql.go @@ -0,0 +1,116 @@ +package postgres + +import ( + "context" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" +) + +const ( + longQueryDuration = 100 * time.Millisecond + longTxnDuration = time.Second // reduce syncing spam +) + +type ( + // A txn wraps a pgx.Tx, logging slow queries. + txn struct { + pgx.Tx + log *zap.Logger + } + + // A row wraps a pgx.Row, logging slow queries. + row struct { + pgx.Row + log *zap.Logger + } + + // rows wraps a pgx.Rows, logging slow queries. + rows struct { + pgx.Rows + log *zap.Logger + } +) + +func (r *rows) Next() bool { + start := time.Now() + next := r.Rows.Next() + if dur := time.Since(start); dur > longQueryDuration { + r.log.Debug("slow next", zap.Duration("elapsed", dur), zap.Stack("stack")) + } + return next +} + +func (r *rows) Scan(dest ...any) error { + start := time.Now() + err := r.Rows.Scan(dest...) + if dur := time.Since(start); dur > longQueryDuration { + r.log.Debug("slow scan", zap.Duration("elapsed", dur), zap.Stack("stack")) + } + return err +} + +func (r *row) Scan(dest ...any) error { + start := time.Now() + err := r.Row.Scan(dest...) + if dur := time.Since(start); dur > longQueryDuration { + r.log.Debug("slow scan", zap.Duration("elapsed", dur), zap.Stack("stack")) + } + return err +} + +// Exec executes a query without returning any rows. The args are for +// any placeholder parameters in the query. +func (tx *txn) Exec(ctx context.Context, query string, args ...any) (pgconn.CommandTag, error) { + start := time.Now() + result, err := tx.Tx.Exec(ctx, query, args...) + if dur := time.Since(start); dur > longQueryDuration { + tx.log.Debug("slow exec", zap.String("query", query), zap.Duration("elapsed", dur), zap.Stack("stack")) + } + return result, err +} + +// Query executes a query that returns rows, typically a SELECT. The +// args are for any placeholder parameters in the query. +func (tx *txn) Query(ctx context.Context, query string, args ...any) (*rows, error) { + start := time.Now() + r, err := tx.Tx.Query(ctx, query, args...) + if dur := time.Since(start); dur > longQueryDuration { + tx.log.Debug("slow query", zap.String("query", query), zap.Duration("elapsed", dur), zap.Stack("stack")) + } + if err != nil { + return nil, err + } + return &rows{r, tx.log.Named("rows")}, nil +} + +// QueryRow executes a query that is expected to return at most one row. +// QueryRow always returns a non-nil value. Errors are deferred until +// Row's Scan method is called. If the query selects no rows, the *Row's +// Scan will return ErrNoRows. Otherwise, the *Row's Scan scans the +// first selected row and discards the rest. +func (tx *txn) QueryRow(ctx context.Context, query string, args ...any) *row { + start := time.Now() + r := tx.Tx.QueryRow(ctx, query, args...) + if dur := time.Since(start); dur > longQueryDuration { + tx.log.Debug("slow query row", zap.String("query", query), zap.Duration("elapsed", dur), zap.Stack("stack")) + } + return &row{r, tx.log.Named("row")} +} + +// getDBVersion returns the current version of the database. +func getDBVersion(ctx context.Context, pool *pgxpool.Pool) (version int64) { + // error is ignored -- the database may not have been initialized yet. + pool.QueryRow(ctx, `SELECT db_version FROM global_settings;`).Scan(&version) + return +} + +// setDBVersion sets the current version of the database. +func setDBVersion(ctx context.Context, tx *txn, version int64) error { + const query = `UPDATE global_settings SET db_version=$1 RETURNING id;` + var dbID int64 + return tx.QueryRow(ctx, query, version).Scan(&dbID) +} diff --git a/persist/postgres/store.go b/persist/postgres/store.go new file mode 100644 index 00000000..94ea84c5 --- /dev/null +++ b/persist/postgres/store.go @@ -0,0 +1,241 @@ +package postgres + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "math" + "math/rand" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + "lukechampine.com/frand" +) + +const ( + factor = 1.8 // factor ^ retryAttempts = backoff time in milliseconds + maxBackoff = 15 * time.Second // max backoff time + maxRetryAttempts = 30 // max number of retry attempts +) + +type ( + // ConnectionInfo contains the information needed to connect to a + // PostgreSQL database. + ConnectionInfo struct { + Host string `json:"host" yaml:"host"` + Port int `json:"port" yaml:"port"` + User string `json:"user" yaml:"user"` + Password string `json:"password" yaml:"password"` + Database string `json:"database" yaml:"database"` + SSLMode string `json:"sslmode" yaml:"sslmode"` + SSLRootCert string `json:"sslrootcert" yaml:"sslrootcert"` + } + + // A Store is a persistent store that uses a PostgreSQL database as its + // backend. + Store struct { + pool *pgxpool.Pool + log *zap.Logger + } +) + +// escapeConnValue escapes a value for use inside a single-quoted libpq +// connection-string parameter. Per the libpq keyword/value syntax, single +// quotes and backslashes within a value must be backslash-escaped. Escaping +// prevents malformed strings and parameter injection for values containing +// these characters (e.g. a password with a single quote). +func escapeConnValue(v string) string { + v = strings.ReplaceAll(v, `\`, `\\`) + v = strings.ReplaceAll(v, `'`, `\'`) + return v +} + +// connString returns the libpq connection string for the given ConnectionInfo. +// It contains the plaintext password and must never be logged; use String for a +// redacted representation. +func (ci ConnectionInfo) connString() string { + params := []string{ + fmt.Sprintf("host='%s'", escapeConnValue(ci.Host)), + fmt.Sprintf("port='%d'", ci.Port), + fmt.Sprintf("user='%s'", escapeConnValue(ci.User)), + fmt.Sprintf("password='%s'", escapeConnValue(ci.Password)), + fmt.Sprintf("dbname='%s'", escapeConnValue(ci.Database)), + } + // only include sslmode when set; an empty sslmode is invalid and differs + // from omitting it (which lets the libpq/pgx default apply). + if ci.SSLMode != "" { + params = append(params, fmt.Sprintf("sslmode='%s'", escapeConnValue(ci.SSLMode))) + } + if ci.SSLRootCert != "" { + params = append(params, fmt.Sprintf("sslrootcert='%s'", escapeConnValue(ci.SSLRootCert))) + } + return strings.Join(params, " ") +} + +// String implements fmt.Stringer. The password is omitted so that accidentally +// logging or formatting a ConnectionInfo does not leak credentials. +func (ci ConnectionInfo) String() string { + return fmt.Sprintf("host=%s port=%d user=%s dbname=%s sslmode=%s", ci.Host, ci.Port, ci.User, ci.Database, ci.SSLMode) +} + +// Close closes the underlying database connection pool. +func (s *Store) Close() error { + s.pool.Close() + return nil +} + +// transaction executes a function within a database transaction. If the +// function returns an error, the transaction is rolled back. Otherwise, the +// transaction is committed. If the transaction fails due to a serialization or +// deadlock error, it is retried up to maxRetryAttempts times before returning. +// +// Because fn may be invoked more than once, it must be idempotent and must not +// rely on side effects outside the transaction (e.g. mutating captured state or +// incrementing counters), as those would be repeated on each retry. +func (s *Store) transaction(fn func(context.Context, *txn) error) error { + var err error + txnID := hex.EncodeToString(frand.Bytes(4)) + log := s.log.Named("transaction").With(zap.String("id", txnID)) + start := time.Now() + for attempt := 1; attempt <= maxRetryAttempts; attempt++ { + attemptStart := time.Now() + log := log.With(zap.Int("attempt", attempt)) + err = s.doTransaction(log, fn) + if err == nil { + // no error, transaction succeeded + return nil + } + + // return immediately if the error is not retryable. non-pg errors are + // never retried, as they may be context cancellations or other + // unexpected errors. + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || !isRetryablePgError(pgErr.Code) { + return fmt.Errorf("transaction failed (attempt %d): %w", attempt, err) + } + + // don't back off after the final attempt + if attempt == maxRetryAttempts { + break + } + + // exponential backoff + sleep := min(time.Duration(math.Pow(factor, float64(attempt)))*time.Millisecond, maxBackoff) + log.Debug("retryable database error", zap.Duration("elapsed", time.Since(attemptStart)), zap.Duration("totalElapsed", time.Since(start)), zap.Duration("retry", sleep), zap.Error(err)) + time.Sleep(sleep + time.Duration(rand.Int63n(int64(sleep/2)))) + } + return fmt.Errorf("transaction failed after %d attempts: %w", maxRetryAttempts, err) +} + +// isRetryablePgError reports whether a PostgreSQL error code represents a +// transient failure that is worth retrying. +func isRetryablePgError(code string) bool { + switch code { + case "40001", // serialization_failure + "40P01", // deadlock_detected + "55P03": // lock_not_available + return true + default: + return false + } +} + +// doTransaction is a helper function to execute a function within a +// transaction. If fn returns an error, the transaction is rolled back. +// Otherwise, the transaction is committed. +func (s *Store) doTransaction(log *zap.Logger, fn func(context.Context, *txn) error) error { + ctx := context.Background() + tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + + failed := true + start := time.Now() + defer func() { + rollbackErr := tx.Rollback(ctx) + if rollbackErr != nil && !errors.Is(rollbackErr, pgx.ErrTxClosed) { + log.Error("failed to rollback transaction", zap.Error(rollbackErr)) + } + // log the transaction if it took longer than txn duration + if time.Since(start) > longTxnDuration { + log.Debug("long transaction", zap.Duration("elapsed", time.Since(start)), zap.Stack("stack"), zap.Bool("failed", failed)) + } + }() + + if err := fn(ctx, &txn{tx, log}); err != nil { + return err + } else if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + + failed = false + return nil +} + +// ensureDatabase creates the configured database if it does not already exist. +func ensureDatabase(ctx context.Context, ci ConnectionInfo) error { + // return early if we're connecting to the default database + if ci.Database == "postgres" { + return nil + } + db := ci.Database + ci.Database = "postgres" + + // connect to the postgres database + pool, err := pgxpool.New(ctx, ci.connString()) + if err != nil { + return fmt.Errorf("failed to connect to postgres database: %w", err) + } + defer pool.Close() + + // check if the database exists + var exists bool + if err := pool.QueryRow(ctx, "SELECT EXISTS(SELECT FROM pg_database WHERE datname = $1)", db).Scan(&exists); err != nil { + return fmt.Errorf("failed to check if database exists: %w", err) + } else if exists { + return nil + } + + // create the database if it does not exist + query := "CREATE DATABASE " + pgx.Identifier{db}.Sanitize() + if _, err := pool.Exec(ctx, query); err != nil { + return fmt.Errorf("failed to create database: %w", err) + } + return nil +} + +// OpenDatabase creates a new PostgreSQL store and initializes the database. If +// the configured database does not exist, it is created. The passed-in context +// is used to establish the connection pool and to ensure the database exists; it +// does not cancel schema initialization or migrations, which run on a background +// context. +func OpenDatabase(ctx context.Context, ci ConnectionInfo, log *zap.Logger) (*Store, error) { + if err := ensureDatabase(ctx, ci); err != nil { + return nil, fmt.Errorf("failed to ensure database %q exists: %w", ci.Database, err) + } + + pool, err := pgxpool.New(ctx, ci.connString()) + if err != nil { + return nil, fmt.Errorf("failed to create pool: %w", err) + } else if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, fmt.Errorf("failed to connect to database: %w", err) + } + + store := &Store{ + pool: pool, + log: log, + } + if err := store.init(int64(len(migrations) + 1)); err != nil { + pool.Close() + return nil, err + } + log.Debug("database initialized", zap.String("database", ci.Database), zap.String("host", ci.Host), zap.Int("port", ci.Port), zap.Int("schemaVersion", len(migrations)+1)) + return store, nil +}