diff --git a/docs/configuration.md b/docs/configuration.md index d4f1b0c7cc..3cf5821d9e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -587,6 +587,39 @@ to `_` (params still apply when provided; short-code overrid --- +### Optional: token.storage.maxPayloadSize + +Maximum serialised size, in bytes, accepted by a single transaction-store write +(`AddTokenRequest`, `AddTransaction`, `AddMovement`). Writes whose payload exceeds +this size are rejected before reaching the database. A value of `0` disables the +check. When the key is absent, the store default (4 MiB) applies. + +```yaml +token: + storage: + maxPayloadSize: 4194304 # 4 MiB +``` + +--- + +### Optional: token.storage.maxPageSize + +Maximum number of rows a single transaction-store read may return. It bounds the +pages accepted by transaction queries and caps token-request queries so an +unbounded scan cannot monopolise database resources. When the key is absent, the +store default (1000) applies. + +```yaml +token: + storage: + maxPageSize: 1000 +``` + +See [Transaction Store API Bounds](services/storage/api-bounds.md) for details, +including why movement/balance queries are intentionally not row-capped. + +--- + ### Optional: token.tms..services.storage.cleanup If not specified, the default configuration is: diff --git a/docs/services/storage.md b/docs/services/storage.md index 6add986ea3..c3c7baa25a 100644 --- a/docs/services/storage.md +++ b/docs/services/storage.md @@ -314,3 +314,10 @@ The cleanup service supports both PostgreSQL and SQLite backends, with different Cleanup behavior is controlled by the configuration section. See the [Configuration Guide](../configuration.md) for detailed parameter descriptions and tuning recommendations. See the [Configuration Guide](../configuration.md), Section `Optional: token.tms..services.network.fabric.recovery`, for detailed parameter descriptions and tuning recommendations. + +## Transaction Store API Bounds + +The transaction store enforces upper bounds on its write and read paths (maximum +serialised write payload, mandatory bounded pagination on transaction queries, +and a hard cap on token-request queries) so a single oversized write or unbounded +scan cannot monopolise database resources. See [**Transaction Store API Bounds**](storage/api-bounds.md). diff --git a/docs/services/storage/api-bounds.md b/docs/services/storage/api-bounds.md new file mode 100644 index 0000000000..410f340669 --- /dev/null +++ b/docs/services/storage/api-bounds.md @@ -0,0 +1,75 @@ +# Transaction Store API Bounds + +The transaction store (shared by **TTXDB** and **AuditDB**) enforces upper bounds +on its write and read paths so that a single oversized write or an unbounded +table scan cannot monopolise database resources. The checks live at the single +SQL implementation in `token/services/storage/db/sql/common/transactions.go`, +which is the only implementor of the write/read interfaces, so both the owner +(TTXDB) and auditor (AuditDB) stores are covered. + +## Write payload limit + +`AddTokenRequest`, `AddTransaction`, and `AddMovement` reject a call whose +serialised payload exceeds a maximum size (in bytes): + +- **`AddTokenRequest`** — measures the raw token request plus the marshalled + application and public metadata (`len(tr) + len(applicationMetadata) + len(publicMetadata)`). +- **`AddTransaction`** / **`AddMovement`** — measure the summed byte length of each + record's fields (transaction/movement IDs, enrollment/sender/recipient IDs, + token type, and the decimal amount). + +The limit defaults to `DefaultMaxPayloadSize` (4 MiB). A value of `0` disables the +check. It can be set from the node configuration under `token.storage.maxPayloadSize` +(bytes), or programmatically with the `WithMaxPayloadSize` store option (same +pattern as `auditdb.WithLocker`): + +```yaml +token: + storage: + maxPayloadSize: 4194304 # 4 MiB; 0 disables the check +``` + +```go +store, _ := common.NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi, + common.WithMaxPayloadSize(8<<20)) // 8 MiB +``` + +## Read result caps + +- **`QueryTransactions`** requires a bounded pagination argument. A `nil` + pagination and the unbounded `pagination.None()` are rejected; an offset page + whose size exceeds the store maximum is also rejected. Callers that need the + full result set page through it (see `pagination.Offset` and + `PageIterator.Pagination.Next`). +- **`QueryTokenRequests`** applies a hard `LIMIT` equal to the store maximum page + size, bounding the scan without changing its signature. +- **`QueryMovements`** is intentionally **not** capped. It feeds balance + aggregation (`HoldingsFilter.Sum`, `PaymentsFilter.Sum`), where truncating rows + would silently corrupt balances. It is already constrained by its required + filter predicates (enrollment IDs, token types, statuses). + +The read cap defaults to `DefaultMaxPageSize` (1000 rows). It can be set from the +node configuration under `token.storage.maxPageSize`, or programmatically with +the `WithMaxPageSize` store option: + +```yaml +token: + storage: + maxPageSize: 1000 +``` + +```go +store, _ := common.NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi, + common.WithMaxPageSize(500)) +``` + +## Notes + +- When a configuration key is absent, the store default applies. Both keys are + read once at driver construction (`LoadStorageConfig`) and applied to the owner + (TTXDB) and auditor (AuditDB) transaction stores in the SQLite and PostgreSQL + drivers. +- The write payload measurement for transaction/movement records uses summed + field lengths rather than a full serialisation, keeping the write hot path + cheap; those records are internally derived, so the meaningful oversized-write + vector is the raw token request handled by `AddTokenRequest`. diff --git a/integration/token/fungible/views/history.go b/integration/token/fungible/views/history.go index 24a07d7491..4971f77f10 100644 --- a/integration/token/fungible/views/history.go +++ b/integration/token/fungible/views/history.go @@ -17,11 +17,44 @@ import ( "github.com/LFDT-Panurus/panurus/token/services/ttx" token2 "github.com/LFDT-Panurus/panurus/token/token" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" + driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver" "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/assert" "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators" "github.com/hyperledger-labs/fabric-smart-client/platform/view/view" ) +// historyPageSize bounds each page when listing all transactions; the storage +// layer rejects unbounded queries, so these views page through the full set. +const historyPageSize = 100 + +// collectAllTransactions pages through queryPage (one page per call) and returns +// every record. queryPage runs the underlying query for the given pagination. +func collectAllTransactions(queryPage func(driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error)) ([]*ttxdb.TransactionRecord, error) { + var page driver2.Pagination + page, err := pagination.Offset(0, historyPageSize) + if err != nil { + return nil, errors.Wrapf(err, "failed to create pagination") + } + var all []*ttxdb.TransactionRecord + for { + items, err := queryPage(page) + if err != nil { + return nil, errors.Wrapf(err, "failed querying transactions") + } + records, err := iterators.ReadAllPointers(items) + if err != nil { + return nil, errors.Wrapf(err, "failed reading transactions") + } + all = append(all, records...) + if len(records) < historyPageSize { + return all, nil + } + if page, err = page.Next(); err != nil { + return nil, errors.Wrapf(err, "failed advancing pagination") + } + } +} + // ListIssuedTokens contains the input to query the list of issued tokens type ListIssuedTokens struct { // Wallet whose identities own the token @@ -143,12 +176,14 @@ func (p *ListAuditedTransactionsView) Call(context view.Context) (any, error) { return nil, errors.Wrapf(err, "failed to get auditor instance") } - it, err := auditor.Transactions(context.Context(), ttxdb.QueryTransactionsParams{From: p.From, To: p.To, SearchDirection: p.SearchDirection}, pagination.None()) - if err != nil { - return nil, errors.Wrapf(err, "failed querying transactions") - } + return collectAllTransactions(func(page driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error) { + it, err := auditor.Transactions(context.Context(), ttxdb.QueryTransactionsParams{From: p.From, To: p.To, SearchDirection: p.SearchDirection}, page) + if err != nil { + return nil, err + } - return iterators.ReadAllPointers(it.Items) + return it.Items, nil + }) } type ListAuditedTransactionsViewFactory struct{} @@ -184,21 +219,24 @@ func (p *ListAcceptedTransactionsView) Call(context view.Context) (any, error) { tms, err := token.GetManagementService(context, ServiceOpts(p.TMSID)...) assert.NoError(err, "failed getting management service") owner := ttx.NewOwner(context, tms) - it, err := owner.Transactions(context.Context(), ttxdb.QueryTransactionsParams{ - SenderWallet: p.SenderWallet, - RecipientWallet: p.RecipientWallet, - From: p.From, - To: p.To, - ActionTypes: p.ActionTypes, - Statuses: p.Statuses, - IDs: p.IDs, - SearchDirection: p.SearchDirection, - }, pagination.None()) - if err != nil { - return nil, errors.Wrapf(err, "failed querying transactions") - } - return iterators.ReadAllPointers(it.Items) + return collectAllTransactions(func(page driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error) { + it, err := owner.Transactions(context.Context(), ttxdb.QueryTransactionsParams{ + SenderWallet: p.SenderWallet, + RecipientWallet: p.RecipientWallet, + From: p.From, + To: p.To, + ActionTypes: p.ActionTypes, + Statuses: p.Statuses, + IDs: p.IDs, + SearchDirection: p.SearchDirection, + }, page) + if err != nil { + return nil, err + } + + return it.Items, nil + }) } type ListAcceptedTransactionsViewFactory struct{} diff --git a/token/services/storage/db/common/checks.go b/token/services/storage/db/common/checks.go index 8e356db6d5..04e5c40479 100644 --- a/token/services/storage/db/common/checks.go +++ b/token/services/storage/db/common/checks.go @@ -15,6 +15,7 @@ import ( "github.com/LFDT-Panurus/panurus/token/services/logging" "github.com/LFDT-Panurus/panurus/token/services/network" "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" + "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/pagination" "github.com/LFDT-Panurus/panurus/token/services/tokens" "github.com/LFDT-Panurus/panurus/token/services/utils" token2 "github.com/LFDT-Panurus/panurus/token/token" @@ -27,6 +28,10 @@ var ( logger = logging.MustGetLogger() ) +// transactionsCheckPageSize is the page size used to iterate all transactions +// during integrity checks; kept below the storage layer's max page size. +const transactionsCheckPageSize = 100 + type TokenTransactionDB interface { GetTokenRequest(ctx context.Context, txID string) ([]byte, error) Transactions(ctx context.Context, params driver.QueryTransactionsParams, pagination driver2.Pagination) (*driver2.PageIterator[*driver.TransactionRecord], error) @@ -115,52 +120,77 @@ func (a *DefaultCheckers) CheckTransactions(ctx context.Context) ([]string, erro return nil, errors.WithMessagef(err, "failed to get ledger [%s]", tms.ID()) } - it, err := a.db.Transactions(ctx, driver.QueryTransactionsParams{}, nil) + // Iterate over transactions one bounded page at a time. A single unbounded + // query is rejected by the storage layer, so we page through the whole set + // using offset pagination and stop once a page comes back short. + var page driver2.Pagination + page, err = pagination.Offset(0, transactionsCheckPageSize) if err != nil { - return nil, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID()) + return nil, errors.WithMessagef(err, "failed to create pagination [%s]", tms.ID()) } - defer it.Items.Close() for { - transactionRecord, err := it.Items.Next() - if err != nil { - return nil, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID()) - } - if transactionRecord == nil { - break - } - - tokenRequest, err := a.db.GetTokenRequest(ctx, transactionRecord.TxID) - if err != nil { - return nil, errors.WithMessagef(err, "failed getting token request [%s]", transactionRecord.TxID) - } - if tokenRequest == nil { - return nil, errors.Errorf("token request [%s] is nil", transactionRecord.TxID) - } - - // check the ledger - lVC, _, err := l.Status(transactionRecord.TxID) - if err != nil { - lVC = network.Unknown - } - switch { - case transactionRecord.Status == driver.Confirmed && lVC != network.Valid: + count, err := func() (int, error) { + it, err := a.db.Transactions(ctx, driver.QueryTransactionsParams{}, page) if err != nil { - errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err)) + return 0, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID()) } - errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is valid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) - case transactionRecord.Status == driver.Deleted && lVC != network.Invalid: - if lVC != network.Unknown || transactionRecord.Status != driver.Deleted { + defer it.Items.Close() + count := 0 + for { + transactionRecord, err := it.Items.Next() + if err != nil { + return 0, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID()) + } + if transactionRecord == nil { + break + } + count++ + + tokenRequest, err := a.db.GetTokenRequest(ctx, transactionRecord.TxID) + if err != nil { + return 0, errors.WithMessagef(err, "failed getting token request [%s]", transactionRecord.TxID) + } + if tokenRequest == nil { + return 0, errors.Errorf("token request [%s] is nil", transactionRecord.TxID) + } + + // check the ledger + lVC, _, err := l.Status(transactionRecord.TxID) if err != nil { - errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err)) + lVC = network.Unknown + } + switch { + case transactionRecord.Status == driver.Confirmed && lVC != network.Valid: + if err != nil { + errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err)) + } + errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is valid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) + case transactionRecord.Status == driver.Deleted && lVC != network.Invalid: + if lVC != network.Unknown || transactionRecord.Status != driver.Deleted { + if err != nil { + errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err)) + } + errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is invalid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) + } + case transactionRecord.Status == driver.Unknown && lVC != network.Unknown: + errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is unknown for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) + case transactionRecord.Status == driver.Pending && lVC == network.Busy: + // this is fine, let's continue + case transactionRecord.Status == driver.Pending && lVC != network.Unknown: + errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is busy for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) } - errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is invalid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) } - case transactionRecord.Status == driver.Unknown && lVC != network.Unknown: - errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is unknown for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) - case transactionRecord.Status == driver.Pending && lVC == network.Busy: - // this is fine, let's continue - case transactionRecord.Status == driver.Pending && lVC != network.Unknown: - errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is busy for vault but not for the ledger [%d]", transactionRecord.TxID, lVC)) + + return count, nil + }() + if err != nil { + return nil, err + } + if count < transactionsCheckPageSize { + break + } + if page, err = page.Next(); err != nil { + return nil, errors.WithMessagef(err, "failed advancing pagination [%s]", tms.ID()) } } diff --git a/token/services/storage/db/dbtest/transactions.go b/token/services/storage/db/dbtest/transactions.go index 52f23cfbe4..d2e8571d32 100644 --- a/token/services/storage/db/dbtest/transactions.go +++ b/token/services/storage/db/dbtest/transactions.go @@ -17,6 +17,7 @@ import ( "github.com/LFDT-Panurus/panurus/token" driver2 "github.com/LFDT-Panurus/panurus/token/driver" driver3 "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" + "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/pagination" "github.com/LFDT-Panurus/panurus/token/services/utils" "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators" "github.com/stretchr/testify/assert" @@ -293,7 +294,9 @@ func TTransaction(t *testing.T, db driver3.TokenTransactionStore) { // get all except last year's t1 := time.Now().Add(time.Second * 3) - it, err := db.QueryTransactions(ctx, driver3.QueryTransactionsParams{From: &t0, To: &t1, SearchDirection: driver3.FromBeginning}, nil) + page, err := pagination.Offset(0, 100) + require.NoError(t, err) + it, err := db.QueryTransactions(ctx, driver3.QueryTransactionsParams{From: &t0, To: &t1, SearchDirection: driver3.FromBeginning}, page) require.NoError(t, err) for _, exp := range txs { act, err := it.Items.Next() @@ -304,7 +307,9 @@ func TTransaction(t *testing.T, db driver3.TokenTransactionStore) { // get all tx from before the first yesterday := t0.AddDate(0, 0, -1).Local().UTC().Truncate(time.Second) - it, err = db.QueryTransactions(ctx, driver3.QueryTransactionsParams{To: &yesterday}, nil) + page, err = pagination.Offset(0, 100) + require.NoError(t, err) + it, err = db.QueryTransactions(ctx, driver3.QueryTransactionsParams{To: &yesterday}, page) require.NoError(t, err) defer it.Items.Close() @@ -851,7 +856,9 @@ func TTransactionQueries(t *testing.T, db driver3.TokenTransactionStore) { func getTransactions(t *testing.T, db driver3.TokenTransactionStore, params driver3.QueryTransactionsParams) []*driver3.TransactionRecord { t.Helper() - records, err := db.QueryTransactions(t.Context(), params, nil) + page, err := pagination.Offset(0, 100) + require.NoError(t, err) + records, err := db.QueryTransactions(t.Context(), params, page) require.NoError(t, err) txs, err := iterators.ReadAllPointers(records.Items) require.NoError(t, err) diff --git a/token/services/storage/db/sql/common/config.go b/token/services/storage/db/sql/common/config.go index a3a59f6874..d14cd925b3 100644 --- a/token/services/storage/db/sql/common/config.go +++ b/token/services/storage/db/sql/common/config.go @@ -38,6 +38,14 @@ const ( // storage: // skipPrefix: true ConfigKeySkipPrefix = "token.storage.skipPrefix" + + // ConfigKeyMaxPayloadSize is the config key for the max write size in bytes + // (0 disables). Absent means use the store default. + ConfigKeyMaxPayloadSize = "token.storage.maxPayloadSize" + + // ConfigKeyMaxPageSize is the config key for the max rows a single read may + // return. Absent means use the store default. + ConfigKeyMaxPageSize = "token.storage.maxPageSize" ) // TableNamesConfig maps a short code (e.g. "id_signers") to the replacement short code @@ -55,6 +63,25 @@ type StorageConfig struct { // SkipPrefix disables the FSC-generated prefix on all table names when true. // Default is false. SkipPrefix bool + // MaxPayloadSize is the max write size in bytes; nil uses the store default, + // non-nil 0 disables the check. + MaxPayloadSize *int + // MaxPageSize is the max rows a single read may return; nil uses the store default. + MaxPageSize *int +} + +// TransactionStoreOptions returns store options for the limits set in config; +// unset limits (nil) keep the store defaults. +func (c StorageConfig) TransactionStoreOptions() []TransactionStoreOption { + var opts []TransactionStoreOption + if c.MaxPayloadSize != nil { + opts = append(opts, WithMaxPayloadSize(*c.MaxPayloadSize)) + } + if c.MaxPageSize != nil { + opts = append(opts, WithMaxPageSize(*c.MaxPageSize)) + } + + return opts } // LoadTableNamesConfig reads the table name overrides from cfg. @@ -87,8 +114,33 @@ func LoadStorageConfig(cfg driver2.Config) (StorageConfig, error) { } } + maxPayloadSize, err := loadOptionalInt(cfg, ConfigKeyMaxPayloadSize) + if err != nil { + return StorageConfig{}, err + } + maxPageSize, err := loadOptionalInt(cfg, ConfigKeyMaxPageSize) + if err != nil { + return StorageConfig{}, err + } + return StorageConfig{ - TableNames: tableNames, - SkipPrefix: skipPrefix, + TableNames: tableNames, + SkipPrefix: skipPrefix, + MaxPayloadSize: maxPayloadSize, + MaxPageSize: maxPageSize, }, nil } + +// loadOptionalInt reads an int config value, returning nil when cfg is nil or +// the key is absent so callers can tell "unset" from a set 0. +func loadOptionalInt(cfg driver2.Config, key string) (*int, error) { + if cfg == nil || !cfg.IsSet(key) { + return nil, nil + } + var v int + if err := cfg.UnmarshalKey(key, &v); err != nil { + return nil, err + } + + return &v, nil +} diff --git a/token/services/storage/db/sql/common/config_test.go b/token/services/storage/db/sql/common/config_test.go index 21042fb6bf..91553f0989 100644 --- a/token/services/storage/db/sql/common/config_test.go +++ b/token/services/storage/db/sql/common/config_test.go @@ -96,6 +96,46 @@ func TestLoadStorageConfigSkipPrefixTrue(t *testing.T) { assert.True(t, result.SkipPrefix) } +// TestLoadStorageConfigLimitsAbsent checks that the payload/page limits are nil +// (unset) when their keys are absent, so store defaults are preserved, and that +// no store options are produced. +func TestLoadStorageConfigLimitsAbsent(t *testing.T) { + cfg := &mockConfig{isSet: false} + result, err := common.LoadStorageConfig(cfg) + require.NoError(t, err) + assert.Nil(t, result.MaxPayloadSize) + assert.Nil(t, result.MaxPageSize) + assert.Empty(t, result.TransactionStoreOptions()) +} + +// TestLoadStorageConfigLimitsPresent checks that the payload/page limits are +// read when set (including an explicit 0, which disables the payload check) and +// that they translate into store options. +func TestLoadStorageConfigLimitsPresent(t *testing.T) { + cfg := &mockConfig{ + isSet: true, + unmarshal: func(key string, rawVal any) error { + switch key { + case common.ConfigKeyMaxPayloadSize: + *rawVal.(*int) = 0 + case common.ConfigKeyMaxPageSize: + *rawVal.(*int) = 250 + case common.ConfigKeyTableNames: + *rawVal.(*common.TableNamesConfig) = common.TableNamesConfig{} + } + + return nil + }, + } + result, err := common.LoadStorageConfig(cfg) + require.NoError(t, err) + require.NotNil(t, result.MaxPayloadSize) + assert.Equal(t, 0, *result.MaxPayloadSize) + require.NotNil(t, result.MaxPageSize) + assert.Equal(t, 250, *result.MaxPageSize) + assert.Len(t, result.TransactionStoreOptions(), 2) +} + // TestLoadStorageConfigTableNamesAndSkipPrefix checks that both fields are // populated correctly when both keys are present. func TestLoadStorageConfigTableNamesAndSkipPrefix(t *testing.T) { diff --git a/token/services/storage/db/sql/common/transactions.go b/token/services/storage/db/sql/common/transactions.go index 085aa02d88..386ca01993 100644 --- a/token/services/storage/db/sql/common/transactions.go +++ b/token/services/storage/db/sql/common/transactions.go @@ -23,6 +23,7 @@ import ( q "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query" common3 "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/common" "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/cond" + paginationutil "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/pagination" _select "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/select" "github.com/hashicorp/go-uuid" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" @@ -37,6 +38,12 @@ import ( // 255 is used as a conservative safe upper bound. const maxAmountBits = 255 +// DefaultMaxPayloadSize is the default max write payload size in bytes (0 disables). +const DefaultMaxPayloadSize = 4 << 20 // 4 MiB + +// DefaultMaxPageSize is the default max rows a single read may return. +const DefaultMaxPageSize = 1000 + type transactionTables struct { Movements string Transactions string @@ -54,6 +61,23 @@ type TransactionStore struct { pi common3.PagInterpreter notifier dbdriver.TransactionNotifier recoveryLeaderFactory func(context.Context, *sql.DB, int64) (dbdriver.RecoveryLeadership, bool, error) + // maxPayloadSize is the max write size in bytes (0 disables the check). + maxPayloadSize int + // maxPageSize is the max rows a single read may return. + maxPageSize int +} + +// TransactionStoreOption configures a TransactionStore at construction time. +type TransactionStoreOption func(*TransactionStore) + +// WithMaxPayloadSize sets the max write payload size in bytes (0 disables). +func WithMaxPayloadSize(n int) TransactionStoreOption { + return func(s *TransactionStore) { s.maxPayloadSize = n } +} + +// WithMaxPageSize sets the max rows a single read may return. +func WithMaxPageSize(n int) TransactionStoreOption { + return func(s *TransactionStore) { s.maxPageSize = n } } func newTransactionStore( @@ -65,8 +89,9 @@ func newTransactionStore( pi common3.PagInterpreter, notifier dbdriver.TransactionNotifier, recoveryLeaderFactory func(context.Context, *sql.DB, int64) (dbdriver.RecoveryLeadership, bool, error), + opts ...TransactionStoreOption, ) *TransactionStore { - return &TransactionStore{ + s := &TransactionStore{ readDB: readDB, writeDB: writeDB, table: tables, @@ -76,7 +101,14 @@ func newTransactionStore( pi: pi, notifier: notifier, recoveryLeaderFactory: recoveryLeaderFactory, + maxPayloadSize: DefaultMaxPayloadSize, + maxPageSize: DefaultMaxPageSize, + } + for _, o := range opts { + o(s) } + + return s } // PrefixedTableName returns the formatted table name for the given logical name. @@ -89,17 +121,17 @@ func (s *TransactionStore) PrefixedTableName(name string) string { return nc.MustFormat(name, s.tableParams...) } -func NewAuditTransactionStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.CondInterpreter, pi common3.PagInterpreter) (*TransactionStore, error) { - return NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi) +func NewAuditTransactionStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.CondInterpreter, pi common3.PagInterpreter, opts ...TransactionStoreOption) (*TransactionStore, error) { + return NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi, opts...) } -func NewOwnerTransactionStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.CondInterpreter, pi common3.PagInterpreter) (*TransactionStore, error) { +func NewOwnerTransactionStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.CondInterpreter, pi common3.PagInterpreter, opts ...TransactionStoreOption) (*TransactionStore, error) { return newTransactionStore(readDB, writeDB, tables.Prefix, tables.Params, transactionTables{ Movements: tables.Movements, Transactions: tables.Transactions, Requests: tables.Requests, TransactionEndorseAck: tables.TransactionEndorseAck, - }, ci, pi, nil, nil), nil + }, ci, pi, nil, nil, opts...), nil } func NewTransactionStoreWithNotifierAndRecovery( @@ -109,13 +141,14 @@ func NewTransactionStoreWithNotifierAndRecovery( pi common3.PagInterpreter, notifier dbdriver.TransactionNotifier, recoveryLeaderFactory func(context.Context, *sql.DB, int64) (dbdriver.RecoveryLeadership, bool, error), + opts ...TransactionStoreOption, ) (*TransactionStore, error) { return newTransactionStore(readDB, writeDB, tables.Prefix, tables.Params, transactionTables{ Movements: tables.Movements, Transactions: tables.Transactions, Requests: tables.Requests, TransactionEndorseAck: tables.TransactionEndorseAck, - }, ci, pi, notifier, recoveryLeaderFactory), nil + }, ci, pi, notifier, recoveryLeaderFactory, opts...), nil } func (db *TransactionStore) CreateSchema() error { @@ -202,6 +235,9 @@ func (db *TransactionStore) QueryMovements(ctx context.Context, params dbdriver. } func (db *TransactionStore) QueryTransactions(ctx context.Context, params dbdriver.QueryTransactionsParams, pagination driver3.Pagination) (*driver3.PageIterator[*dbdriver.TransactionRecord], error) { + if err := paginationutil.ValidateBounded(pagination, db.maxPageSize); err != nil { + return nil, err + } transactionsTable, requestsTable := q.Table(db.table.Transactions), q.Table(db.table.Requests) query, args := q.Select(). Fields( @@ -277,12 +313,14 @@ func (db *TransactionStore) Notifier() (dbdriver.TransactionNotifier, error) { return db.notifier, nil } -// QueryTokenRequests returns an iterator over the token requests matching the passed params +// QueryTokenRequests returns an iterator over the token requests matching params, +// capped at the store's max page size to bound the scan. func (db *TransactionStore) QueryTokenRequests(ctx context.Context, params dbdriver.QueryTokenRequestsParams) (dbdriver.TokenRequestIterator, error) { query, args := q.Select(). FieldsByName("tx_id", "request", "status"). From(q.Table(db.table.Requests)). Where(cond.In("status", params.Statuses...)). + Limit(db.maxPageSize). Format(db.ci) logging.Debug(logger, query, args) @@ -501,9 +539,10 @@ func (db *TransactionStore) NewTransactionStoreTransaction() (dbdriver.Transacti } return &TransactionStoreTransaction{ - txn: txn, - table: &db.table, - ci: db.ci, + txn: txn, + table: &db.table, + ci: db.ci, + maxPayloadSize: db.maxPayloadSize, }, nil } @@ -511,6 +550,8 @@ type TransactionStoreTransaction struct { txn *sql.Tx table *transactionTables ci common3.CondInterpreter + // maxPayloadSize is the max write size in bytes (0 disables the check). + maxPayloadSize int } func (w *TransactionStoreTransaction) Impl() dbdriver.TransactionImpl { @@ -538,11 +579,21 @@ func (w *TransactionStoreTransaction) Rollback() { w.txn = nil } +// checkPayloadSize rejects a write whose size exceeds maxPayloadSize (when > 0). +func (w *TransactionStoreTransaction) checkPayloadSize(op string, size int) error { + if w.maxPayloadSize > 0 && size > w.maxPayloadSize { + return errors.Errorf("%s payload size %d bytes exceeds maximum %d bytes", op, size, w.maxPayloadSize) + } + + return nil +} + func (w *TransactionStoreTransaction) AddTransaction(ctx context.Context, rs ...dbdriver.TransactionRecord) error { if w.txn == nil { return errors.New("no db transaction in progress") } rows := make([]common3.Tuple, len(rs)) + payloadSize := 0 for i, r := range rs { logger.DebugfContext(ctx, "adding transaction record [%s:%d,%s:%s:%s:%s]", r.TxID, r.ActionType, r.TokenType, r.SenderEID, r.RecipientEID, r.Amount) id, err := uuid.GenerateUUID() @@ -552,8 +603,12 @@ func (w *TransactionStoreTransaction) AddTransaction(ctx context.Context, rs ... if r.Amount.BitLen() > maxAmountBits { return errors.Errorf("amount [%s] exceeds maximum supported size of %d bits", r.Amount, maxAmountBits) } + payloadSize += len(r.TxID) + len(r.SenderEID) + len(r.RecipientEID) + len(r.TokenType) + len(r.Amount.String()) rows[i] = common3.Tuple{id, r.TxID, int(r.ActionType), r.SenderEID, r.RecipientEID, r.TokenType, r.Amount.String(), r.Timestamp.UTC()} } + if err := w.checkPayloadSize("AddTransaction", payloadSize); err != nil { + return err + } query, args := q.InsertInto(w.table.Transactions). Fields("id", "tx_id", "action_type", "sender_eid", "recipient_eid", "token_type", "amount", "stored_at"). @@ -585,6 +640,9 @@ func (w *TransactionStoreTransaction) AddTokenRequest(ctx context.Context, txID if err != nil { return errors.New("error marshaling application metadata") } + if err := w.checkPayloadSize("AddTokenRequest", len(tr)+len(ja)+len(jp)); err != nil { + return err + } query, args := q.InsertInto(w.table.Requests). Fields("tx_id", "request", "status", "status_message", "application_metadata", "public_metadata", "pp_hash", "stored_at"). @@ -609,6 +667,7 @@ func (w *TransactionStoreTransaction) AddMovement(ctx context.Context, rs ...dbd now := time.Now().UTC() rows := make([]common3.Tuple, len(rs)) + payloadSize := 0 for i, r := range rs { logger.DebugfContext(ctx, "adding movement record [%s]", r) id, err := uuid.GenerateUUID() @@ -618,8 +677,12 @@ func (w *TransactionStoreTransaction) AddMovement(ctx context.Context, rs ...dbd if r.Amount.BitLen() > maxAmountBits { return errors.Errorf("amount [%s] exceeds maximum supported size of %d bits", r.Amount, maxAmountBits) } + payloadSize += len(r.TxID) + len(r.EnrollmentID) + len(r.TokenType) + len(r.Amount.String()) rows[i] = common3.Tuple{id, r.TxID, r.EnrollmentID, r.TokenType, r.Amount.String(), now} } + if err := w.checkPayloadSize("AddMovement", payloadSize); err != nil { + return err + } query, args := q.InsertInto(w.table.Movements). Fields("id", "tx_id", "enrollment_id", "token_type", "amount", "stored_at"). diff --git a/token/services/storage/db/sql/common/transactions_bounds_test.go b/token/services/storage/db/sql/common/transactions_bounds_test.go new file mode 100644 index 0000000000..410e0730c7 --- /dev/null +++ b/token/services/storage/db/sql/common/transactions_bounds_test.go @@ -0,0 +1,101 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package common + +import ( + "database/sql" + "math/big" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" + "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/pagination" + "github.com/LFDT-Panurus/panurus/token/token" + "github.com/onsi/gomega" +) + +func boundsTestStore(db *sql.DB, opts ...TransactionStoreOption) *TransactionStore { + store, _ := NewOwnerTransactionStore(db, db, TableNames{ + Movements: "MOVEMENTS", + Transactions: "TRANSACTIONS", + Requests: "REQUESTS", + TransactionEndorseAck: "TRANSACTION_ENDORSE_ACK", + }, nil, nil, opts...) + + return store +} + +// TestWritePayloadSizeLimit verifies that each write method rejects a payload +// that exceeds the configured maximum before touching the database. +func TestWritePayloadSizeLimit(t *testing.T) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + store := boundsTestStore(db, WithMaxPayloadSize(20)) + mockDB.ExpectBegin() + w, err := store.NewTransactionStoreTransaction() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + big100 := strings.Repeat("a", 100) + + // AddTokenRequest: raw request alone exceeds the limit. + err = w.AddTokenRequest(t.Context(), "tx", make([]byte, 100), nil, nil, nil) + gomega.Expect(err).To(gomega.HaveOccurred()) + gomega.Expect(err.Error()).To(gomega.ContainSubstring("exceeds maximum")) + + // AddTransaction: summed field lengths exceed the limit. + err = w.AddTransaction(t.Context(), driver.TransactionRecord{ + TxID: "tx", SenderEID: big100, RecipientEID: "bob", + TokenType: token.Type("USD"), Amount: big.NewInt(1), + }) + gomega.Expect(err).To(gomega.HaveOccurred()) + gomega.Expect(err.Error()).To(gomega.ContainSubstring("exceeds maximum")) + + // AddMovement: summed field lengths exceed the limit. + err = w.AddMovement(t.Context(), driver.MovementRecord{ + TxID: "tx", EnrollmentID: big100, + TokenType: token.Type("USD"), Amount: big.NewInt(1), + }) + gomega.Expect(err).To(gomega.HaveOccurred()) + gomega.Expect(err.Error()).To(gomega.ContainSubstring("exceeds maximum")) +} + +// TestWritePayloadSizeDisabled verifies that a zero limit disables the check: +// a large payload is accepted (the writes proceed to the DB). +func TestWritePayloadSizeDisabled(t *testing.T) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + store := boundsTestStore(db, WithMaxPayloadSize(0)) + mockDB.ExpectBegin() + mockDB.ExpectExec("INSERT INTO REQUESTS").WillReturnResult(sqlmock.NewResult(1, 1)) + w, err := store.NewTransactionStoreTransaction() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + err = w.AddTokenRequest(t.Context(), "tx", make([]byte, 100), nil, nil, nil) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + gomega.Expect(mockDB.ExpectationsWereMet()).To(gomega.Succeed()) +} + +// TestQueryTransactionsRejectsUnbounded verifies the read guard rejects nil and +// unbounded (None) pagination without querying the database. +func TestQueryTransactionsRejectsUnbounded(t *testing.T) { + gomega.RegisterTestingT(t) + db, _, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + store := boundsTestStore(db) + + _, err = store.QueryTransactions(t.Context(), driver.QueryTransactionsParams{}, nil) + gomega.Expect(err).To(gomega.HaveOccurred()) + + _, err = store.QueryTransactions(t.Context(), driver.QueryTransactionsParams{}, pagination.None()) + gomega.Expect(err).To(gomega.HaveOccurred()) +} diff --git a/token/services/storage/db/sql/common/transactions_test_util.go b/token/services/storage/db/sql/common/transactions_test_util.go index cb84adfd93..8804af60da 100644 --- a/token/services/storage/db/sql/common/transactions_test_util.go +++ b/token/services/storage/db/sql/common/transactions_test_util.go @@ -121,10 +121,12 @@ func TestQueryTransactions(t *testing.T, store transactionsStoreConstructor) { "FROM TRANSACTIONS LEFT JOIN REQUESTS ON TRANSACTIONS.tx_id = REQUESTS.tx_id ORDER BY TRANSACTIONS.stored_at DESC"). WillReturnRows(mockDB.NewRows([]string{"tx_id", "action_type", "sender_eid", "recipient_eid", "token_type", "amount", "status", "application_metadata", "public_metadata", "stored_at"}).AddRow(output...)) + page, err := pagination.Offset(0, 10) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) info, err := store(db).QueryTransactions(t.Context(), driver.QueryTransactionsParams{ IDs: []string{}, - }, pagination.None()) + }, page) gomega.Expect(mockDB.ExpectationsWereMet()).To(gomega.Succeed()) gomega.Expect(err).ToNot(gomega.HaveOccurred()) @@ -174,8 +176,8 @@ func TestQueryTokenRequests(t *testing.T, store transactionsStoreConstructor, tr statusClause = "\\(\\(status = \\$1\\)\\) OR \\(\\(status = \\$2\\)\\)" } mockDB. - ExpectQuery("SELECT tx_id, request, status FROM REQUESTS WHERE "+statusClause). - WithArgs(driver.Deleted, driver.Unknown). + ExpectQuery("SELECT tx_id, request, status FROM REQUESTS WHERE "+statusClause+" LIMIT \\$3"). + WithArgs(driver.Deleted, driver.Unknown, DefaultMaxPageSize). WillReturnRows(mockDB.NewRows([]string{"tx_id", "request", "status"}).AddRow(output...)) it, err := store(db).QueryTokenRequests(t.Context(), diff --git a/token/services/storage/db/sql/postgres/driver.go b/token/services/storage/db/sql/postgres/driver.go index 668ca17f04..2a3a5d6383 100644 --- a/token/services/storage/db/sql/postgres/driver.go +++ b/token/services/storage/db/sql/postgres/driver.go @@ -58,10 +58,12 @@ func NewDriver(config driver3.Config) *Driver { // NewDriverWithDbProvider returns a new Driver for Postgres using the given database provider. func NewDriverWithDbProvider(config driver3.Config, dbProvider fscPostgres.DbProvider) *Driver { - tableNamesConfig, err := common3.LoadTableNamesConfig(config) + storageConfig, err := common3.LoadStorageConfig(config) if err != nil { - logger.Warnf("failed to load table name overrides: %v — using defaults", err) + logger.Warnf("failed to load storage config: %v — using defaults", err) } + tableNamesConfig := storageConfig.TableNames + storageOpts := storageConfig.TransactionStoreOptions() d := &Driver{ cp: fscPostgres.NewConfigProvider(common.NewConfig(config)), @@ -72,8 +74,10 @@ func NewDriverWithDbProvider(config driver3.Config, dbProvider fscPostgres.DbPro d.Wallet = newProviderWithKeyMapper(dbProvider, NewWalletStore, "wallet", tableNamesConfig) d.Identity = newIdentityStoreProvider(dbProvider, tableNamesConfig) d.Token = newTokenStoreProvider(dbProvider, tableNamesConfig) - d.AuditTx = newProviderWithKeyMapper(dbProvider, NewAuditTransactionStore, "audittx", tableNamesConfig) - d.OwnerTx = newTransactionStoreProvider(dbProvider, tableNamesConfig) + d.AuditTx = newProviderWithKeyMapper(dbProvider, func(dbs *common.RWDB, tableNames common3.TableNames) (*AuditTransactionStore, error) { + return NewAuditTransactionStore(dbs, tableNames, storageOpts...) + }, "audittx", tableNamesConfig) + d.OwnerTx = newTransactionStoreProvider(dbProvider, tableNamesConfig, storageOpts...) d.Endorser = newEndorserStoreProvider(dbProvider, tableNamesConfig) d.KeyStore = newProviderWithKeyMapper(dbProvider, NewKeystoreStore, "keystore", tableNamesConfig) @@ -172,7 +176,7 @@ func newIdentityStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesConfi } // newTransactionStoreProvider returns a lazy provider for TransactionStore with notifier support. -func newTransactionStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesConfig common3.TableNamesConfig) lazy.Provider[fscPostgres.Config, *TransactionStore] { +func newTransactionStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesConfig common3.TableNamesConfig, storeOpts ...common3.TransactionStoreOption) lazy.Provider[fscPostgres.Config, *TransactionStore] { return lazy.NewProviderWithKeyMapper(key, func(o fscPostgres.Config) (*TransactionStore, error) { opts := fscPostgres.Opts{ DataSource: o.DataSource, @@ -199,7 +203,7 @@ func newTransactionStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesCo } // db - p, err := NewTransactionStoreWithNotifier(dbs, tableNames, notifier) + p, err := NewTransactionStoreWithNotifier(dbs, tableNames, notifier, storeOpts...) if err != nil { return nil, err } diff --git a/token/services/storage/db/sql/postgres/transactions.go b/token/services/storage/db/sql/postgres/transactions.go index 78cc92e8b3..f3f022d9ce 100644 --- a/token/services/storage/db/sql/postgres/transactions.go +++ b/token/services/storage/db/sql/postgres/transactions.go @@ -68,7 +68,7 @@ func (s *TransactionStore) CreateSchema() error { } // NewTransactionStoreWithNotifier creates a new TransactionStore with the provided notifier and recovery support. -func NewTransactionStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, notifier *TransactionNotifier) (*TransactionStore, error) { +func NewTransactionStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, notifier *TransactionNotifier, opts ...sqlcommon.TransactionStoreOption) (*TransactionStore, error) { // Create recovery leader factory using PostgreSQL advisory locks recoveryLeaderFactory := NewAdvisoryLockFactory() @@ -80,6 +80,7 @@ func NewTransactionStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.Tab NewPaginationInterpreter(), notifier, recoveryLeaderFactory, + opts..., ) if err != nil { return nil, err @@ -95,13 +96,14 @@ func NewTransactionStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.Tab } // NewAuditTransactionStore creates a new AuditTransactionStore. -func NewAuditTransactionStore(dbs *scommon.RWDB, tableNames sqlcommon.TableNames) (*AuditTransactionStore, error) { +func NewAuditTransactionStore(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, opts ...sqlcommon.TransactionStoreOption) (*AuditTransactionStore, error) { baseStore, err := sqlcommon.NewAuditTransactionStore( dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter(), NewPaginationInterpreter(), + opts..., ) if err != nil { return nil, err diff --git a/token/services/storage/db/sql/query/pagination/validate.go b/token/services/storage/db/sql/query/pagination/validate.go new file mode 100644 index 0000000000..013edcb911 --- /dev/null +++ b/token/services/storage/db/sql/query/pagination/validate.go @@ -0,0 +1,37 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package pagination + +import ( + "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" + "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver" +) + +// ValidateBounded rejects pagination that would run an unbounded scan: nil, +// None(), or an offset whose page size is non-positive or exceeds maxPageSize +// (capped only when maxPageSize > 0). Empty and keyset pagination are allowed. +func ValidateBounded(p driver.Pagination, maxPageSize int) error { + switch v := p.(type) { + case nil: + return errors.New("pagination is required: a bounded page must be provided") + case *none: + return errors.New("unbounded pagination (None) is not allowed") + case *empty: + return nil + case *offset: + if v.PageSize <= 0 { + return errors.Errorf("pagination page size must be positive, got %d", v.PageSize) + } + if maxPageSize > 0 && v.PageSize > maxPageSize { + return errors.Errorf("pagination page size %d exceeds maximum %d", v.PageSize, maxPageSize) + } + + return nil + default: + return nil + } +} diff --git a/token/services/storage/db/sql/query/pagination/validate_test.go b/token/services/storage/db/sql/query/pagination/validate_test.go new file mode 100644 index 0000000000..3dc7a2a0a6 --- /dev/null +++ b/token/services/storage/db/sql/query/pagination/validate_test.go @@ -0,0 +1,49 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package pagination + +import "testing" + +func TestValidateBounded(t *testing.T) { + off, err := Offset(0, 10) + if err != nil { + t.Fatalf("unexpected error building offset: %v", err) + } + bigOff, err := Offset(0, 100) + if err != nil { + t.Fatalf("unexpected error building offset: %v", err) + } + zeroOff, err := Offset(0, 0) + if err != nil { + t.Fatalf("unexpected error building offset: %v", err) + } + + cases := []struct { + name string + build func() error + wantErr bool + }{ + {"nil is rejected", func() error { return ValidateBounded(nil, 50) }, true}, + {"None is rejected", func() error { return ValidateBounded(None(), 50) }, true}, + {"Empty is allowed", func() error { return ValidateBounded(Empty(), 50) }, false}, + {"bounded offset is allowed", func() error { return ValidateBounded(off, 50) }, false}, + {"offset over max is rejected", func() error { return ValidateBounded(bigOff, 50) }, true}, + {"zero page size is rejected", func() error { return ValidateBounded(zeroOff, 50) }, true}, + {"no max disables the cap", func() error { return ValidateBounded(bigOff, 0) }, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + err := c.build() + if c.wantErr && err == nil { + t.Fatalf("expected an error, got nil") + } + if !c.wantErr && err != nil { + t.Fatalf("expected no error, got %v", err) + } + }) + } +} diff --git a/token/services/storage/db/sql/sqlite/driver.go b/token/services/storage/db/sql/sqlite/driver.go index ba5311f006..90deff1e6f 100644 --- a/token/services/storage/db/sql/sqlite/driver.go +++ b/token/services/storage/db/sql/sqlite/driver.go @@ -53,10 +53,12 @@ func NewDriver(config driver3.Config) *Driver { } func NewDriverWithDbProvider(config driver3.Config, dbProvider fscSqlite.DbProvider) *Driver { - tableNamesConfig, err := common2.LoadTableNamesConfig(config) + storageConfig, err := common2.LoadStorageConfig(config) if err != nil { - logger.Warnf("failed to load table name overrides: %v — using defaults", err) + logger.Warnf("failed to load storage config: %v — using defaults", err) } + tableNamesConfig := storageConfig.TableNames + storageOpts := storageConfig.TransactionStoreOptions() d := &Driver{ cp: fscSqlite.NewConfigProvider(common.NewConfig(config)), @@ -67,8 +69,12 @@ func NewDriverWithDbProvider(config driver3.Config, dbProvider fscSqlite.DbProvi d.Wallet = newProviderWithKeyMapper(dbProvider, NewWalletStore, tableNamesConfig) d.Identity = newIdentityStoreProvider(dbProvider, tableNamesConfig) d.Token = newProviderWithKeyMapper(dbProvider, NewTokenStore, tableNamesConfig) - d.AuditTx = newProviderWithKeyMapper(dbProvider, NewAuditTransactionStore, tableNamesConfig) - d.OwnerTx = newProviderWithKeyMapper(dbProvider, NewTransactionStore, tableNamesConfig) + d.AuditTx = newProviderWithKeyMapper(dbProvider, func(dbs *common.RWDB, tableNames common2.TableNames) (*AuditTransactionStore, error) { + return NewAuditTransactionStore(dbs, tableNames, storageOpts...) + }, tableNamesConfig) + d.OwnerTx = newProviderWithKeyMapper(dbProvider, func(dbs *common.RWDB, tableNames common2.TableNames) (*OwnerTransactionStore, error) { + return NewTransactionStore(dbs, tableNames, storageOpts...) + }, tableNamesConfig) d.Endorser = newProviderWithKeyMapper(dbProvider, NewEndorserStore, tableNamesConfig) d.KeyStore = newProviderWithKeyMapper(dbProvider, NewKeystoreStore, tableNamesConfig) diff --git a/token/services/storage/db/sql/sqlite/transactions.go b/token/services/storage/db/sql/sqlite/transactions.go index 1d6f1766f8..2ea1ec5b60 100644 --- a/token/services/storage/db/sql/sqlite/transactions.go +++ b/token/services/storage/db/sql/sqlite/transactions.go @@ -18,10 +18,10 @@ type ( OwnerTransactionStore = common3.TransactionStore ) -func NewAuditTransactionStore(dbs *common2.RWDB, tableNames common3.TableNames) (*AuditTransactionStore, error) { - return common3.NewAuditTransactionStore(dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter(), pagination.NewDefaultInterpreter()) +func NewAuditTransactionStore(dbs *common2.RWDB, tableNames common3.TableNames, opts ...common3.TransactionStoreOption) (*AuditTransactionStore, error) { + return common3.NewAuditTransactionStore(dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter(), pagination.NewDefaultInterpreter(), opts...) } -func NewTransactionStore(dbs *common2.RWDB, tableNames common3.TableNames) (*OwnerTransactionStore, error) { - return common3.NewOwnerTransactionStore(dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter(), pagination.NewDefaultInterpreter()) +func NewTransactionStore(dbs *common2.RWDB, tableNames common3.TableNames, opts ...common3.TransactionStoreOption) (*OwnerTransactionStore, error) { + return common3.NewOwnerTransactionStore(dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter(), pagination.NewDefaultInterpreter(), opts...) }