diff --git a/ocp/balance/calculator.go b/ocp/balance/calculator.go index 0d13e25..73343cc 100644 --- a/ocp/balance/calculator.go +++ b/ocp/balance/calculator.go @@ -395,6 +395,93 @@ func defaultBatchCalculationFromCache(ctx context.Context, data ocp_data.Provide return res, nil } +// BalanceWithUsdCostBasis holds a token account's quark balance and USD cost +// basis, in balance.UsdQuarksPerUnit. +type BalanceWithUsdCostBasis struct { + Quarks uint64 + UsdCostBasis int64 +} + +// BatchCalculateWithUsdCostBasisFromCache calculates balances and USD cost +// bases for a set of account records. When ledger reads are enabled, both +// values for an account come from the same balance record read, so they are +// guaranteed consistent with each other. Accounts without a backfilled +// record fall back to the legacy aggregates, which read the two values from +// separate sources. +// +// Note: Use this method when calculating balances for accounts that are managed by +// Code (ie. Timelock account) and operate within the L2 system. +func BatchCalculateWithUsdCostBasisFromCache(ctx context.Context, data ocp_data.Provider, accountRecordsBatch ...*common.AccountRecords) (map[string]*BalanceWithUsdCostBasis, error) { + tracer := metrics.TraceMethodCall(ctx, metricsPackageName, "BatchCalculateWithUsdCostBasisFromCache") + defer tracer.End() + + var tokenAccounts []string + for _, accountRecords := range accountRecordsBatch { + if !accountRecords.IsTimelock() || !accountRecords.IsManagedByCode(ctx) { + tracer.OnError(ErrNotManagedByCode) + return nil, ErrNotManagedByCode + } + tokenAccounts = append(tokenAccounts, accountRecords.General.TokenAccount) + } + + balanceRecords := make(map[string]*balance.Record) + if enableLedgerReads.Get(ctx) { + var err error + balanceRecords, err = data.GetBalanceBatch(ctx, tokenAccounts...) + if err != nil { + tracer.OnError(err) + return nil, err + } + } + + res := make(map[string]*BalanceWithUsdCostBasis, len(tokenAccounts)) + var remaining []string + for _, tokenAccount := range tokenAccounts { + balanceRecord, ok := balanceRecords[tokenAccount] + if !ok || !balanceRecord.IsBackfilled { + remaining = append(remaining, tokenAccount) + continue + } + + quarks, err := quarksFromRecord(balanceRecord) + if err != nil { + tracer.OnError(err) + return nil, err + } + res[tokenAccount] = &BalanceWithUsdCostBasis{ + Quarks: quarks, + UsdCostBasis: balanceRecord.UsdCostBasis, + } + } + + if len(remaining) == 0 { + return res, nil + } + + legacyQuarks, err := CalculateBatch( + ctx, + remaining, + NetBalanceFromIntentActionsBatch(ctx, data), + FundingFromExternalDepositsBatch(ctx, data), + ) + if err != nil { + tracer.OnError(err) + return nil, err + } + for _, tokenAccount := range remaining { + usdCostBasis, err := legacyUsdCostBasis(ctx, data, tokenAccount) + if err != nil { + tracer.OnError(err) + return nil, err + } + res[tokenAccount] = &BalanceWithUsdCostBasis{ + Quarks: legacyQuarks[tokenAccount], + UsdCostBasis: usdCostBasis, + } + } + return res, nil +} + // CalculateUsdCostBasisFromCache calculates a token account's USD cost basis, // in balance.UsdQuarksPerUnit, using cached values. // diff --git a/ocp/balance/calculator_test.go b/ocp/balance/calculator_test.go index fc6ff0a..a03223d 100644 --- a/ocp/balance/calculator_test.go +++ b/ocp/balance/calculator_test.go @@ -419,6 +419,73 @@ func TestDefaultCalculationMethods_BalanceRecord(t *testing.T) { assert.Equal(t, expected, balanceByAccount) } +func TestDefaultCalculationMethods_BalanceWithUsdCostBasis(t *testing.T) { + env := setupBalanceTestEnv(t) + enableLedgerReadsForTest(t) + + vmConfig := testutil.NewRandomVmConfig(t, true) + backfilledOwner := testutil.NewRandomAccount(t) + backfilledAccount, err := backfilledOwner.ToTimelockVault(vmConfig) + require.NoError(t, err) + legacyOwner := testutil.NewRandomAccount(t) + legacyAccount, err := legacyOwner.ToTimelockVault(vmConfig) + require.NoError(t, err) + + externalAccount := testutil.NewRandomAccount(t) + + data := &balanceTestData{ + vmConfig: vmConfig, + codeUsers: []*common.Account{backfilledOwner, legacyOwner}, + transactions: []balanceTestTransaction{ + {source: externalAccount, destination: backfilledAccount, quantity: 11, transactionState: transaction.ConfirmationFinalized}, + {source: externalAccount, destination: legacyAccount, quantity: 33, transactionState: transaction.ConfirmationFinalized}, + }, + } + + setupBalanceTestData(t, env, data) + + // Both values come from the same record for a backfilled account, even + // where it disagrees with history + require.NoError(t, env.data.CreateBalance(env.ctx, &balance.Record{ + TokenAccount: backfilledAccount.PublicKey().ToBase58(), + OwnerAccount: backfilledOwner.PublicKey().ToBase58(), + MintAccount: vmConfig.Mint.PublicKey().ToBase58(), + Quarks: 42, + UsdCostBasis: 4_200_000, + IsOpen: true, + IsLocked: true, + IsBackfilled: true, + })) + + accountRecordsBatch := make([]*common.AccountRecords, 0) + for _, tokenAccount := range []*common.Account{backfilledAccount, legacyAccount} { + generalRecord, err := env.data.GetAccountInfoByTokenAddress(env.ctx, tokenAccount.PublicKey().ToBase58()) + require.NoError(t, err) + timelockRecord, err := env.data.GetTimelockByVault(env.ctx, tokenAccount.PublicKey().ToBase58()) + require.NoError(t, err) + accountRecordsBatch = append(accountRecordsBatch, &common.AccountRecords{ + General: generalRecord, + Timelock: timelockRecord, + }) + } + + res, err := BatchCalculateWithUsdCostBasisFromCache(env.ctx, env.data, accountRecordsBatch...) + require.NoError(t, err) + require.Len(t, res, 2) + + cached := res[backfilledAccount.PublicKey().ToBase58()] + require.NotNil(t, cached) + assert.EqualValues(t, 42, cached.Quarks) + assert.EqualValues(t, 4_200_000, cached.UsdCostBasis) + + // An account without a backfilled record falls back to the legacy + // aggregates for both values + cached = res[legacyAccount.PublicKey().ToBase58()] + require.NotNil(t, cached) + assert.EqualValues(t, 33, cached.Quarks) + assert.EqualValues(t, 0, cached.UsdCostBasis) // deposits aren't primary-owner intents in this fixture +} + func TestDefaultCalculationMethods_UnlockedBalanceRecord(t *testing.T) { env := setupBalanceTestEnv(t) enableLedgerReadsForTest(t) @@ -471,6 +538,7 @@ func TestDefaultCalculationMethods_UnlockedBalanceRecord(t *testing.T) { func TestDefaultCalculationMethods_BalanceRecordReadsDisabled(t *testing.T) { env := setupBalanceTestEnv(t) + disableLedgerReadsForTest(t) vmConfig := testutil.NewRandomVmConfig(t, true) owner := testutil.NewRandomAccount(t) @@ -603,6 +671,14 @@ func enableLedgerReadsForTest(t *testing.T) { }) } +func disableLedgerReadsForTest(t *testing.T) { + previous := enableLedgerReads + enableLedgerReads = wrapper.NewBoolConfig(memory.NewConfig(false), defaultEnableLedgerReads) + t.Cleanup(func() { + enableLedgerReads = previous + }) +} + type balanceTestEnv struct { ctx context.Context data ocp_data.Provider diff --git a/ocp/balance/config.go b/ocp/balance/config.go index 78be8a4..6925291 100644 --- a/ocp/balance/config.go +++ b/ocp/balance/config.go @@ -15,8 +15,8 @@ const ( // to the ledger at all. When disabled, it is a no-op. EnableLedgerWritesConfigEnvName = "BALANCE_ENABLE_LEDGER_WRITES" - defaultEnableLedgerReads = false - defaultEnableLedgerWrites = false + defaultEnableLedgerReads = true + defaultEnableLedgerWrites = true ) var ( diff --git a/ocp/balance/ledger.go b/ocp/balance/ledger.go index 99f7fa0..4ad81c5 100644 --- a/ocp/balance/ledger.go +++ b/ocp/balance/ledger.go @@ -14,6 +14,12 @@ import ( // ledger doesn't track. var ErrUntrackedAccount = errors.New("account is not tracked by the balance ledger") +// LedgerReadsEnabled reports whether backfilled ledger records are the +// authoritative source for balance reads. +func LedgerReadsEnabled(ctx context.Context) bool { + return enableLedgerReads.Get(ctx) +} + // LedgerWritesEnabled reports whether the ledger is being written to. // Callers use it to skip building deltas entirely when writes are disabled, // since builders reject flows the ledger doesn't support. diff --git a/ocp/balance/ledger_test.go b/ocp/balance/ledger_test.go index 5f72e57..5bf6caa 100644 --- a/ocp/balance/ledger_test.go +++ b/ocp/balance/ledger_test.go @@ -20,6 +20,7 @@ import ( func TestApplyDeltasInTx_WritesDisabled(t *testing.T) { ctx := context.Background() data := ocp_data.NewTestDataProvider() + disableLedgerWritesForTest(t) source := newLedgerTestAccount(t, ctx, data, commonpb.AccountType_PRIMARY) @@ -153,9 +154,12 @@ func TestCreateRecordInTx(t *testing.T) { // Disabled writes are a no-op primary := newLedgerTestAccountInfo(t, ctx, data, commonpb.AccountType_PRIMARY) - require.NoError(t, CreateRecordInTx(ctx, data, primary)) - _, err := data.GetBalance(ctx, primary.TokenAccount) - assert.Equal(t, balance.ErrRecordNotFound, err) + func() { + disableLedgerWritesForTest(t) + require.NoError(t, CreateRecordInTx(ctx, data, primary)) + _, err := data.GetBalance(ctx, primary.TokenAccount) + assert.Equal(t, balance.ErrRecordNotFound, err) + }() enableLedgerWritesForTest(t) @@ -210,6 +214,14 @@ func newLedgerTestAccountInfo(t *testing.T, ctx context.Context, data ocp_data.P return record } +func disableLedgerWritesForTest(t *testing.T) { + previous := enableLedgerWrites + enableLedgerWrites = wrapper.NewBoolConfig(memory.NewConfig(false), defaultEnableLedgerWrites) + t.Cleanup(func() { + enableLedgerWrites = previous + }) +} + func enableLedgerWritesForTest(t *testing.T) { previous := enableLedgerWrites enableLedgerWrites = wrapper.NewBoolConfig(memory.NewConfig(true), defaultEnableLedgerWrites) diff --git a/ocp/balance/lock.go b/ocp/balance/lock.go deleted file mode 100644 index c38e156..0000000 --- a/ocp/balance/lock.go +++ /dev/null @@ -1,75 +0,0 @@ -package balance - -import ( - "context" - "errors" - - "github.com/code-payments/ocp-server/ocp/common" - ocp_data "github.com/code-payments/ocp-server/ocp/data" - "github.com/code-payments/ocp-server/ocp/data/balance" -) - -// OptimisticVersionLock is an optimistic version lock on an account's cached -// balance, which can be paired with DB updates against balances that need to -// be protected against race conditions. -type OptimisticVersionLock struct { - vault *common.Account - currentVersion uint64 -} - -// GetOptimisticVersionLock gets an optimistic version lock for the vault account's -// cached balance -func GetOptimisticVersionLock(ctx context.Context, data ocp_data.Provider, vault *common.Account) (*OptimisticVersionLock, error) { - version, err := data.GetCachedBalanceVersion(ctx, vault.PublicKey().ToBase58()) - if err != nil { - return nil, err - } - return &OptimisticVersionLock{ - vault: vault, - currentVersion: version, - }, nil -} - -// OnNewBalanceVersion is called in the DB transaction updating the account's -// cached balance -func (l *OptimisticVersionLock) OnNewBalanceVersion(ctx context.Context, data ocp_data.Provider) error { - return data.AdvanceCachedBalanceVersion(ctx, l.vault.PublicKey().ToBase58(), l.currentVersion) -} - -// RequireSameBalanceVerion is called in the DB transaction requireing the -// account's cached balance not be changed -func (l *OptimisticVersionLock) RequireSameBalanceVerion(ctx context.Context, data ocp_data.Provider) error { - latestVersion, err := data.GetCachedBalanceVersion(ctx, l.vault.PublicKey().ToBase58()) - if err != nil { - return err - } - if latestVersion < l.currentVersion { - return errors.New("unexpected balance version detected") - } - if l.currentVersion != latestVersion { - return balance.ErrStaleCachedBalanceVersion - } - return nil -} - -// OpenCloseStatusLock is a lock on an account's open/close status -type OpenCloseStatusLock struct { - vault *common.Account -} - -func NewOpenCloseStatusLock(vault *common.Account) *OpenCloseStatusLock { - return &OpenCloseStatusLock{ - vault: vault, - } -} - -// OnPaymentToAccount is called in the DB transaction making a payment to the -// account that may be closed -func (l *OpenCloseStatusLock) OnPaymentToAccount(ctx context.Context, data ocp_data.Provider) error { - return data.CheckNotClosedForBalanceUpdate(ctx, l.vault.PublicKey().ToBase58()) -} - -// OnClose is called in the DB transaction closing the account -func (l *OpenCloseStatusLock) OnClose(ctx context.Context, data ocp_data.Provider) error { - return data.MarkAsClosedForBalanceUpdate(ctx, l.vault.PublicKey().ToBase58()) -} diff --git a/ocp/data/balance/memory/store.go b/ocp/data/balance/memory/store.go index f31e676..179e20c 100644 --- a/ocp/data/balance/memory/store.go +++ b/ocp/data/balance/memory/store.go @@ -14,9 +14,7 @@ type store struct { balanceRecords []*balance.Record balanceRecordsByTokenAccount map[string]*balance.Record - cachedBalanceVersionsByAccount map[string]uint64 - closedAccounts map[string]any - externalCheckpointRecords []*balance.ExternalCheckpointRecord + externalCheckpointRecords []*balance.ExternalCheckpointRecord last uint64 } @@ -24,9 +22,7 @@ type store struct { // New returns a new in memory balance.Store func New() balance.Store { return &store{ - balanceRecordsByTokenAccount: make(map[string]*balance.Record), - cachedBalanceVersionsByAccount: make(map[string]uint64), - closedAccounts: make(map[string]any), + balanceRecordsByTokenAccount: make(map[string]*balance.Record), } } @@ -139,6 +135,20 @@ func (s *store) GetAllLockedByMint(_ context.Context, mint string, minQuarks int return res, nil } +// CountLockedByMint implements balance.Store.CountLockedByMint +func (s *store) CountLockedByMint(_ context.Context, mint string, minQuarks int64) (uint64, error) { + s.mu.Lock() + defer s.mu.Unlock() + + var res uint64 + for _, item := range s.balanceRecordsByTokenAccount { + if item.MintAccount == mint && item.Quarks >= minQuarks && item.IsLocked && item.IsBackfilled { + res++ + } + } + return res, nil +} + // MarkAsUnlocked implements balance.Store.MarkAsUnlocked func (s *store) MarkAsUnlocked(_ context.Context, tokenAccount string) error { s.mu.Lock() @@ -218,6 +228,9 @@ func applyDelta(item *balance.Record, delta *balance.Delta) error { item.Quarks += int64(delta.Quarks) item.UsdCostBasis += delta.UsdCostBasis case balance.DeltaDebit: + if enforce && !item.IsOpen { + return balance.ErrAccountClosed + } if enforce && item.Quarks < int64(delta.Quarks) { return balance.ErrInsufficientBalance } @@ -313,8 +326,6 @@ func (s *store) reset() { s.balanceRecords = nil s.balanceRecordsByTokenAccount = make(map[string]*balance.Record) - s.cachedBalanceVersionsByAccount = make(map[string]uint64) - s.closedAccounts = make(map[string]any) s.externalCheckpointRecords = nil s.last = 0 } diff --git a/ocp/data/balance/memory/store_legacy.go b/ocp/data/balance/memory/store_legacy.go deleted file mode 100644 index 9e062f9..0000000 --- a/ocp/data/balance/memory/store_legacy.go +++ /dev/null @@ -1,66 +0,0 @@ -package memory - -import ( - "context" - - "github.com/code-payments/ocp-server/ocp/data/balance" -) - -// GetCachedVersion implements balance.Store.GetCachedVersion -func (s *store) GetCachedVersion(_ context.Context, account string) (uint64, error) { - s.mu.Lock() - defer s.mu.Unlock() - - current, ok := s.cachedBalanceVersionsByAccount[account] - if !ok { - return 0, nil - } - return current, nil -} - -// AdvanceCachedVersion implements balance.Store.AdvanceCachedVersion -func (s *store) AdvanceCachedVersion(_ context.Context, account string, currentVersion uint64) error { - s.mu.Lock() - defer s.mu.Unlock() - - actualVersion, ok := s.cachedBalanceVersionsByAccount[account] - if !ok { - if currentVersion != 0 { - return balance.ErrStaleCachedBalanceVersion - } - - s.cachedBalanceVersionsByAccount[account] = 1 - - return nil - } - - if actualVersion != currentVersion { - return balance.ErrStaleCachedBalanceVersion - } - - s.cachedBalanceVersionsByAccount[account]++ - - return nil -} - -// CheckNotClosed implements balance.Store.CheckNotClosed -func (s *store) CheckNotClosed(ctx context.Context, account string) error { - s.mu.Lock() - defer s.mu.Unlock() - - if _, ok := s.closedAccounts[account]; ok { - return balance.ErrAccountClosed - } - - return nil -} - -// MarkAsClosed implements balance.Store.MarkAsClosed -func (s *store) MarkAsClosed(ctx context.Context, account string) error { - s.mu.Lock() - defer s.mu.Unlock() - - s.closedAccounts[account] = true - - return nil -} diff --git a/ocp/data/balance/postgres/model.go b/ocp/data/balance/postgres/model.go index dbe59dc..9a29133 100644 --- a/ocp/data/balance/postgres/model.go +++ b/ocp/data/balance/postgres/model.go @@ -180,6 +180,17 @@ func dbGetAllLockedByMint(ctx context.Context, db *sqlx.DB, mint string, minQuar return res, nil } +func dbCountLockedByMint(ctx context.Context, db *sqlx.DB, mint string, minQuarks int64) (uint64, error) { + var res uint64 + query := `SELECT COUNT(*) FROM ` + tableName + ` + WHERE mint_account = $1 AND quarks >= $2 AND is_locked AND is_backfilled` + err := db.GetContext(ctx, &res, query, mint, minQuarks) + if err != nil { + return 0, err + } + return res, nil +} + func dbMarkAsUnlocked(ctx context.Context, db *sqlx.DB, tokenAccount string) error { return pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error { query := `UPDATE ` + tableName + ` @@ -217,7 +228,7 @@ func dbApplyDeltas(ctx context.Context, db *sqlx.DB, deltas []*balance.Delta) er case balance.DeltaDebit: query = `UPDATE ` + tableName + ` SET quarks = quarks - $2, usd_cost_basis = usd_cost_basis - $3, updated_at = $4 - WHERE token_account = $1 AND (NOT is_backfilled OR (is_locked AND quarks >= $2))` + WHERE token_account = $1 AND (NOT is_backfilled OR (is_open AND is_locked AND quarks >= $2))` args = []any{delta.TokenAccount, int64(delta.Quarks), delta.UsdCostBasis, time.Now().UTC()} case balance.DeltaDrain: query = `UPDATE ` + tableName + ` @@ -281,6 +292,9 @@ func classifyFailedDelta(delta *balance.Delta, current *balance.Record) error { } switch delta.Kind { case balance.DeltaDebit: + if !current.IsOpen { + return balance.ErrAccountClosed + } return balance.ErrInsufficientBalance case balance.DeltaDrain, balance.DeltaClose: if !current.IsOpen { diff --git a/ocp/data/balance/postgres/model_legacy.go b/ocp/data/balance/postgres/model_legacy.go deleted file mode 100644 index 247c9a4..0000000 --- a/ocp/data/balance/postgres/model_legacy.go +++ /dev/null @@ -1,121 +0,0 @@ -package postgres - -import ( - "context" - "database/sql" - "errors" - - "github.com/jmoiron/sqlx" - - pgutil "github.com/code-payments/ocp-server/database/postgres" - "github.com/code-payments/ocp-server/ocp/data/balance" -) - -const ( - cachedBalanceVersionTableName = "ocp__core_cachedbalanceversion" - openCloseLocksTableName = "ocp__core_opencloselocks" -) - -func dbGetCachedVersion(ctx context.Context, db *sqlx.DB, account string) (uint64, error) { - var res uint64 - err := pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error { - insertQuery := `INSERT INTO ` + cachedBalanceVersionTableName + ` - (token_account, version) - VALUES($1, 0) - ON CONFLICT DO NOTHING - ` - sqlResult, err := tx.ExecContext(ctx, insertQuery, account) - if err != nil { - return err - } - rowsAffected, err := sqlResult.RowsAffected() - if err != nil { - return err - } - if rowsAffected == 1 { - res = 0 - return nil - } - - selectQuery := `SELECT version FROM ` + cachedBalanceVersionTableName + ` - WHERE token_account = $1 - FOR UPDATE` - return db.GetContext(ctx, &res, selectQuery, account) - }) - return res, err - -} - -func dbAdvanceCachedVersion(ctx context.Context, db *sqlx.DB, account string, currentVersion uint64) error { - return pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error { - var res uint64 - query := `UPDATE ` + cachedBalanceVersionTableName + ` - SET version = version + 1 - WHERE token_account = $1 AND version = $2 - RETURNING version - ` - err := tx.GetContext(ctx, &res, query, account, currentVersion) - if pgutil.IsNoRows(err) || pgutil.IsUniqueViolation(err) { - return balance.ErrStaleCachedBalanceVersion - } - if err != nil { - return err - } - return nil - }) - -} - -func dbCheckNotClosed(ctx context.Context, db *sqlx.DB, account string) error { - return pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error { - insertQuery := `INSERT INTO ` + openCloseLocksTableName + ` - (token_account, is_open) - VALUES ($1, TRUE) - ON CONFLICT DO NOTHING - ` - - _, err := tx.ExecContext(ctx, insertQuery, account) - if err != nil { - return err - } - - selectQuery := `SELECT is_open FROM ` + openCloseLocksTableName + ` - WHERE token_account = $1 - FOR UPDATE - ` - var isOpen bool - err = tx.GetContext(ctx, &isOpen, selectQuery, account) - if err != nil { - return err - } - if !isOpen { - return balance.ErrAccountClosed - } - return nil - }) -} - -func dbMarkAsClosed(ctx context.Context, db *sqlx.DB, account string) error { - return pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error { - query := `INSERT INTO ` + openCloseLocksTableName + ` - (token_account, is_open) - VALUES ($1, FALSE) - - ON CONFLICT (token_account) - DO UPDATE - SET is_open = FALSE - WHERE ` + openCloseLocksTableName + `.token_account = $1 - - RETURNING is_open - ` - var isOpen bool - err := tx.GetContext(ctx, &isOpen, query, account) - if err != nil { - return err - } - if isOpen { - return errors.New("unexpected state transition") - } - return nil - }) -} diff --git a/ocp/data/balance/postgres/store.go b/ocp/data/balance/postgres/store.go index a47a362..67d5b96 100644 --- a/ocp/data/balance/postgres/store.go +++ b/ocp/data/balance/postgres/store.go @@ -86,6 +86,11 @@ func (s *store) GetAllLockedByMint(ctx context.Context, mint string, minQuarks i return fromModels(models), nil } +// CountLockedByMint implements balance.Store.CountLockedByMint +func (s *store) CountLockedByMint(ctx context.Context, mint string, minQuarks int64) (uint64, error) { + return dbCountLockedByMint(ctx, s.db, mint, minQuarks) +} + // MarkAsUnlocked implements balance.Store.MarkAsUnlocked func (s *store) MarkAsUnlocked(ctx context.Context, tokenAccount string) error { return dbMarkAsUnlocked(ctx, s.db, tokenAccount) diff --git a/ocp/data/balance/postgres/store_legacy.go b/ocp/data/balance/postgres/store_legacy.go deleted file mode 100644 index f947bf1..0000000 --- a/ocp/data/balance/postgres/store_legacy.go +++ /dev/null @@ -1,25 +0,0 @@ -package postgres - -import ( - "context" -) - -// GetCachedVersion implements balance.Store.GetCachedVersion -func (s *store) GetCachedVersion(ctx context.Context, account string) (uint64, error) { - return dbGetCachedVersion(ctx, s.db, account) -} - -// AdvanceCachedVersion implements balance.Store.AdvanceCachedVersion -func (s *store) AdvanceCachedVersion(ctx context.Context, account string, currentVersion uint64) error { - return dbAdvanceCachedVersion(ctx, s.db, account, currentVersion) -} - -// CheckNotClosed implements balance.Store.CheckNotClosed -func (s *store) CheckNotClosed(ctx context.Context, account string) error { - return dbCheckNotClosed(ctx, s.db, account) -} - -// MarkAsClosed implements balance.Store.MarkAsClosed -func (s *store) MarkAsClosed(ctx context.Context, account string) error { - return dbMarkAsClosed(ctx, s.db, account) -} diff --git a/ocp/data/balance/postgres/store_test.go b/ocp/data/balance/postgres/store_test.go index 50d1fa0..b713f24 100644 --- a/ocp/data/balance/postgres/store_test.go +++ b/ocp/data/balance/postgres/store_test.go @@ -47,24 +47,6 @@ const ( CREATE INDEX ocp__core_balance__idx__owner_account__mint_account ON ocp__core_balance (owner_account, mint_account); CREATE INDEX ocp__core_balance__idx__mint_account__id ON ocp__core_balance (mint_account, id) WHERE is_locked; - CREATE TABLE ocp__core_cachedbalanceversion ( - id SERIAL NOT NULL PRIMARY KEY, - - token_account TEXT NOT NULL, - version INTEGER NOT NULL, - - CONSTRAINT ocp__core_cachedbalanceversion__unique__token_account UNIQUE (token_account) - ); - - CREATE TABLE ocp__core_opencloselocks ( - id SERIAL NOT NULL PRIMARY KEY, - - token_account TEXT NOT NULL, - is_open BOOL NOT NULL, - - CONSTRAINT ocp__core_opencloselocks__unique__token_account UNIQUE (token_account) - ); - CREATE TABLE ocp__core_externalbalancecheckpoint ( id SERIAL NOT NULL PRIMARY KEY, @@ -81,8 +63,6 @@ const ( // Used for testing ONLY, the table and migrations are external to this repository tableDestroy = ` DROP TABLE ocp__core_balance; - DROP TABLE ocp__core_cachedbalanceversion; - DROP TABLE ocp__core_opencloselocks; DROP TABLE ocp__core_externalbalancecheckpoint; ` ) diff --git a/ocp/data/balance/store.go b/ocp/data/balance/store.go index bf3d693..b427b17 100644 --- a/ocp/data/balance/store.go +++ b/ocp/data/balance/store.go @@ -25,8 +25,6 @@ var ( // reviewed rather than recorded. ErrNegativeBalance = errors.New("backfilled balance is negative") - ErrStaleCachedBalanceVersion = errors.New("cached balance version is stale") - ErrAccountClosed = errors.New("account open state is stale") // ErrAccountUnlocked is returned when a delta other than a credit targets @@ -93,6 +91,13 @@ type Store interface { // ErrRecordNotFound is returned if no records exist. GetAllLockedByMint(ctx context.Context, mint string, minQuarks int64, cursor query.Cursor, limit uint64, direction query.Ordering) ([]*Record, error) + // CountLockedByMint counts locked, backfilled records for a mint with at + // least minQuarks. Records that are not backfilled are excluded, since + // their balances are partial sums that can't be compared against a + // threshold. Unlocked records are excluded, since their balances are + // stale. + CountLockedByMint(ctx context.Context, mint string, minQuarks int64) (uint64, error) + // ApplyDeltas atomically applies a set of deltas. Either every delta is // applied or none are. Deltas are applied in SortDeltas order. // @@ -106,11 +111,12 @@ type Store interface { // // ErrInsufficientBalance is returned when a debit exceeds the balance. // ErrBalanceChanged is returned when a drain or close doesn't match the - // balance. ErrAccountClosed is returned when a credit, drain or close - // targets a closed account. ErrAccountUnlocked is returned when a delta - // other than a credit targets an unlocked account, whose record is no - // longer maintained. DeltaAdjustUsdCostBasis carries no predicate and - // only ever fails when the record is missing. + // balance. ErrAccountClosed is returned when a credit, debit, drain or + // close targets a closed account, which is frozen. ErrAccountUnlocked is + // returned when a delta other than a credit targets an unlocked account, + // whose record is no longer maintained. DeltaAdjustUsdCostBasis carries + // no predicate: it moves no quarks, so it applies to closed and unlocked + // accounts alike and only fails when the record is missing. ApplyDeltas(ctx context.Context, deltas ...*Delta) error // MarkAsUnlocked marks an account's timelock vault as unlocked, which is @@ -142,36 +148,4 @@ type Store interface { // // ErrCheckpointNotFound is returend if no DB record exists. GetExternalCheckpoint(ctx context.Context, account string) (*ExternalCheckpointRecord, error) - - // GetCachedVersion gets the current cached balance version, which can be used - // for optimistic locking cached balances for operations with outgoing transfers. - // - // Note: Use ApplyDeltas, whose predicates replace the version check. - // Retained for accounts that are not yet backfilled. - GetCachedVersion(ctx context.Context, account string) (uint64, error) - - // AdvanceCachedVersion advances an account's cached balance version. - // - // ErrStaleCachedBalanceVersion is returned if the currentVersion is out of date. - // - // Note: Use ApplyDeltas, whose predicates replace the version check. - // Retained for accounts that are not yet backfilled. - AdvanceCachedVersion(ctx context.Context, account string, currentVersion uint64) error - - // CheckNotClosed checks whether an account is closed under a lock to guarantee - // payments to a closeable destination with cached balances are made to an open - // account. - // - // ErrAccountClosed is returned if the account has been closed. - // - // Note: Use ApplyDeltas with DeltaCredit. Retained for accounts that - // are not yet backfilled. - CheckNotClosed(ctx context.Context, account string) error - - // MarkAsClosed marks an account as being closed and unable to receive payments - // as a destination. - // - // Note: Use ApplyDeltas with DeltaDrain or DeltaClose. Retained for - // accounts that are not yet backfilled. - MarkAsClosed(ctx context.Context, account string) error } diff --git a/ocp/data/balance/tests/tests.go b/ocp/data/balance/tests/tests.go index 9823fd0..bb569dd 100644 --- a/ocp/data/balance/tests/tests.go +++ b/ocp/data/balance/tests/tests.go @@ -23,8 +23,6 @@ func RunTests(t *testing.T, s balance.Store, teardown func()) { testApplyDeltasConcurrency, testBackfill, testMarkAsUnlocked, - testCachedBalanceVersionHappyPath, - testClosedAccountHappyPath, testExternalCheckpointHappyPath, } { tf(t, s) @@ -196,6 +194,34 @@ func testGetAllLockedByMint(t *testing.T, s balance.Store) { _, err = s.GetAllLockedByMint(ctx, "mint_1", 0, query.ToCursor(5), 10, query.Ascending) assert.Equal(t, balance.ErrRecordNotFound, err) + + // Counting by mint uses the same threshold semantics, but only over + // backfilled records + require.NoError(t, s.Create(ctx, &balance.Record{ + TokenAccount: "token_account_not_backfilled", + OwnerAccount: "owner", + MintAccount: "mint_1", + Quarks: 1000, + IsOpen: true, + IsLocked: true, + IsBackfilled: false, + })) + + count, err := s.CountLockedByMint(ctx, "mint_1", 0) + require.NoError(t, err) + assert.EqualValues(t, 5, count) + + count, err = s.CountLockedByMint(ctx, "mint_1", 20) + require.NoError(t, err) + assert.EqualValues(t, 3, count) + + count, err = s.CountLockedByMint(ctx, "mint_2", 0) + require.NoError(t, err) + assert.EqualValues(t, 1, count) + + count, err = s.CountLockedByMint(ctx, "mint_3", 0) + require.NoError(t, err) + assert.EqualValues(t, 0, count) }) } @@ -244,6 +270,12 @@ func testApplyDeltasBackfilled(t *testing.T, s balance.Store) { require.NoError(t, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaCredit, UsdCostBasis: -5})) assertBalance(t, s, "token_account_1", 70, -15, true) + // So can a debit, while the account is open + require.NoError(t, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDebit, UsdCostBasis: -5})) + assertBalance(t, s, "token_account_1", 70, -10, true) + require.NoError(t, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDebit, UsdCostBasis: 5})) + assertBalance(t, s, "token_account_1", 70, -15, true) + assert.Equal(t, balance.ErrBalanceChanged, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaClose})) assert.Equal(t, balance.ErrBalanceChanged, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDrain, Quarks: 69})) assert.Equal(t, balance.ErrBalanceChanged, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDrain, Quarks: 71})) @@ -255,7 +287,11 @@ func testApplyDeltasBackfilled(t *testing.T, s balance.Store) { assert.Equal(t, balance.ErrAccountClosed, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaCredit, Quarks: 1})) assert.Equal(t, balance.ErrAccountClosed, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDrain, Quarks: 0, UsdCostBasis: 1})) assert.Equal(t, balance.ErrAccountClosed, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaClose})) - assert.Equal(t, balance.ErrInsufficientBalance, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDebit, Quarks: 1})) + + // A closed account is frozen: even a zero-quark cost basis adjustment + // cannot leave it + assert.Equal(t, balance.ErrAccountClosed, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDebit, Quarks: 1})) + assert.Equal(t, balance.ErrAccountClosed, s.ApplyDeltas(ctx, &balance.Delta{TokenAccount: "token_account_1", Kind: balance.DeltaDebit, UsdCostBasis: 1})) assertBalance(t, s, "token_account_1", 0, 0, false) // A cost basis adjustment carries no predicate, so it still applies to @@ -633,44 +669,6 @@ func testMarkAsUnlocked(t *testing.T, s balance.Store) { }) } -func testCachedBalanceVersionHappyPath(t *testing.T, s balance.Store) { - t.Run("testCachedBalanceVersionHappyPath", func(t *testing.T) { - ctx := context.Background() - - for i := range 100 { - for range 10 { - currentVersion, err := s.GetCachedVersion(ctx, "token_account_1") - require.NoError(t, err) - assert.EqualValues(t, i, currentVersion) - } - - if i > 0 { - assert.Equal(t, balance.ErrStaleCachedBalanceVersion, s.AdvanceCachedVersion(ctx, "token_account_1", uint64(i-1))) - } - assert.Equal(t, balance.ErrStaleCachedBalanceVersion, s.AdvanceCachedVersion(ctx, "token_account_1", uint64(i+1))) - - require.NoError(t, s.AdvanceCachedVersion(ctx, "token_account_1", uint64(i))) - } - - currentVersion, err := s.GetCachedVersion(ctx, "token_account_2") - require.NoError(t, err) - assert.EqualValues(t, 0, currentVersion) - }) -} - -func testClosedAccountHappyPath(t *testing.T, s balance.Store) { - t.Run("testClosedAccountHappyPath", func(t *testing.T) { - ctx := context.Background() - - require.NoError(t, s.CheckNotClosed(ctx, "token_account_1")) - - require.NoError(t, s.MarkAsClosed(ctx, "token_account_1")) - - assert.Equal(t, balance.ErrAccountClosed, s.CheckNotClosed(ctx, "token_account_1")) - require.NoError(t, s.CheckNotClosed(ctx, "token_account_2s")) - }) -} - func testExternalCheckpointHappyPath(t *testing.T, s balance.Store) { t.Run("testExternalCheckpointHappyPath", func(t *testing.T) { ctx := context.Background() diff --git a/ocp/data/internal.go b/ocp/data/internal.go index 4dc7b75..af4a9e0 100644 --- a/ocp/data/internal.go +++ b/ocp/data/internal.go @@ -127,13 +127,10 @@ type DatabaseData interface { GetAllBalancesByOwner(ctx context.Context, owner string) ([]*balance.Record, error) GetAllBalancesByOwnerAndMint(ctx context.Context, owner, mint string) ([]*balance.Record, error) GetAllLockedBalancesByMint(ctx context.Context, mint string, minQuarks int64, cursor query.Cursor, limit uint64, direction query.Ordering) ([]*balance.Record, error) + CountLockedBalancesByMint(ctx context.Context, mint string, minQuarks int64) (uint64, error) ApplyBalanceDeltas(ctx context.Context, deltas ...*balance.Delta) error MarkBalanceAsUnlocked(ctx context.Context, tokenAccount string) error BackfillBalance(ctx context.Context, tokenAccount string, fn balance.BackfillFunc) error - GetCachedBalanceVersion(ctx context.Context, account string) (uint64, error) - AdvanceCachedBalanceVersion(ctx context.Context, account string, currentVersion uint64) error - CheckNotClosedForBalanceUpdate(ctx context.Context, account string) error - MarkAsClosedForBalanceUpdate(ctx context.Context, account string) error SaveExternalBalanceCheckpoint(ctx context.Context, record *balance.ExternalCheckpointRecord) error GetExternalBalanceCheckpoint(ctx context.Context, account string) (*balance.ExternalCheckpointRecord, error) @@ -490,24 +487,15 @@ func (dp *DatabaseProvider) GetAllLockedBalancesByMint(ctx context.Context, mint func (dp *DatabaseProvider) MarkBalanceAsUnlocked(ctx context.Context, tokenAccount string) error { return dp.balance.MarkAsUnlocked(ctx, tokenAccount) } +func (dp *DatabaseProvider) CountLockedBalancesByMint(ctx context.Context, mint string, minQuarks int64) (uint64, error) { + return dp.balance.CountLockedByMint(ctx, mint, minQuarks) +} func (dp *DatabaseProvider) ApplyBalanceDeltas(ctx context.Context, deltas ...*balance.Delta) error { return dp.balance.ApplyDeltas(ctx, deltas...) } func (dp *DatabaseProvider) BackfillBalance(ctx context.Context, tokenAccount string, fn balance.BackfillFunc) error { return dp.balance.Backfill(ctx, tokenAccount, fn) } -func (dp *DatabaseProvider) GetCachedBalanceVersion(ctx context.Context, account string) (uint64, error) { - return dp.balance.GetCachedVersion(ctx, account) -} -func (dp *DatabaseProvider) AdvanceCachedBalanceVersion(ctx context.Context, account string, currentVersion uint64) error { - return dp.balance.AdvanceCachedVersion(ctx, account, currentVersion) -} -func (dp *DatabaseProvider) CheckNotClosedForBalanceUpdate(ctx context.Context, account string) error { - return dp.balance.CheckNotClosed(ctx, account) -} -func (dp *DatabaseProvider) MarkAsClosedForBalanceUpdate(ctx context.Context, account string) error { - return dp.balance.MarkAsClosed(ctx, account) -} func (dp *DatabaseProvider) SaveExternalBalanceCheckpoint(ctx context.Context, record *balance.ExternalCheckpointRecord) error { return dp.balance.SaveExternalCheckpoint(ctx, record) } diff --git a/ocp/rpc/account/server.go b/ocp/rpc/account/server.go index 8ea416d..b5ed848 100644 --- a/ocp/rpc/account/server.go +++ b/ocp/rpc/account/server.go @@ -25,6 +25,7 @@ import ( ocp_data "github.com/code-payments/ocp-server/ocp/data" "github.com/code-payments/ocp-server/ocp/data/account" "github.com/code-payments/ocp-server/ocp/data/action" + balance_store "github.com/code-payments/ocp-server/ocp/data/balance" "github.com/code-payments/ocp-server/ocp/rpc" account_worker "github.com/code-payments/ocp-server/ocp/worker/account" timelock_token_v1 "github.com/code-payments/ocp-server/solana/timelock/v1" @@ -36,8 +37,9 @@ var ( ) type balanceMetadata struct { - value uint64 - source accountpb.TokenAccountInfo_BalanceSource + quarks uint64 + usdCostBasis int64 + source accountpb.TokenAccountInfo_BalanceSource } type server struct { @@ -321,19 +323,21 @@ func (s *server) fetchBalances(ctx context.Context, allAccountRecords []*common. // Don't calculate a balance for now, since the caching strategy // is not possible. balanceMetadataByTokenAccount[accountRecords.General.TokenAccount] = &balanceMetadata{ - value: 0, - source: accountpb.TokenAccountInfo_BALANCE_SOURCE_UNKNOWN, + quarks: 0, + usdCostBasis: 0, + source: accountpb.TokenAccountInfo_BALANCE_SOURCE_UNKNOWN, } } } - balancesByTokenAccount, err := balance.BatchCalculateFromCacheWithAccountRecords(ctx, s.data, mangedByCodeRecords...) + balancesByTokenAccount, err := balance.BatchCalculateWithUsdCostBasisFromCache(ctx, s.data, mangedByCodeRecords...) if err != nil { return nil, err } - for tokenAccount, quarks := range balancesByTokenAccount { + for tokenAccount, cached := range balancesByTokenAccount { balanceMetadataByTokenAccount[tokenAccount] = &balanceMetadata{ - value: quarks, - source: accountpb.TokenAccountInfo_BALANCE_SOURCE_CACHE, + quarks: cached.Quarks, + usdCostBasis: cached.UsdCostBasis, + source: accountpb.TokenAccountInfo_BALANCE_SOURCE_CACHE, } } @@ -372,8 +376,9 @@ func (s *server) fetchBalances(ctx context.Context, allAccountRecords []*common. protoBalanceSource = accountpb.TokenAccountInfo_BALANCE_SOURCE_UNKNOWN } balanceMetadataByTokenAccount[tokenAccount.PublicKey().ToBase58()] = &balanceMetadata{ - value: quarks, - source: protoBalanceSource, + quarks: quarks, + usdCostBasis: 0, + source: protoBalanceSource, } } @@ -453,7 +458,7 @@ func (s *server) getProtoAccountInfo(ctx context.Context, records *common.Accoun } // Otherwise, check whether it looks like the gift card was claimed. - if prefetchedBalanceMetadata.source == accountpb.TokenAccountInfo_BALANCE_SOURCE_CACHE && prefetchedBalanceMetadata.value == 0 { + if prefetchedBalanceMetadata.source == accountpb.TokenAccountInfo_BALANCE_SOURCE_CACHE && prefetchedBalanceMetadata.quarks == 0 { claimState = accountpb.TokenAccountInfo_CLAIM_STATE_CLAIMED } else if records.Timelock.IsClosed() { claimState = accountpb.TokenAccountInfo_CLAIM_STATE_CLAIMED @@ -481,7 +486,8 @@ func (s *server) getProtoAccountInfo(ctx context.Context, records *common.Accoun // If the gift card account is claimed or expired, force the balance to zero. if claimState == accountpb.TokenAccountInfo_CLAIM_STATE_CLAIMED || claimState == accountpb.TokenAccountInfo_CLAIM_STATE_EXPIRED { prefetchedBalanceMetadata.source = accountpb.TokenAccountInfo_BALANCE_SOURCE_CACHE - prefetchedBalanceMetadata.value = 0 + prefetchedBalanceMetadata.quarks = 0 + prefetchedBalanceMetadata.usdCostBasis = 0 } } @@ -495,18 +501,11 @@ func (s *server) getProtoAccountInfo(ctx context.Context, records *common.Accoun var usdCostBasis float64 if common.IsCoreMint(mintAccount) && common.IsCoreMintUsdStableCoin() { - usdCostBasis = float64(prefetchedBalanceMetadata.value) / float64(common.CoreMintQuarksPerUnit) + usdCostBasis = float64(prefetchedBalanceMetadata.quarks) / float64(common.CoreMintQuarksPerUnit) } else { - switch records.General.AccountType { - case commonpb.AccountType_PRIMARY: - // todo: Assumes the structure that each user has exactly one primary account per mint - usdCostBasis, err = s.data.GetUsdCostBasis(ctx, ownerAccount.PublicKey().ToBase58(), mintAccount.PublicKey().ToBase58()) - if err != nil { - return nil, err - } - default: - usdCostBasis = 0 // Account type not supported - } + // Prefetched alongside the balance, from the same record where the + // ledger serves both + usdCostBasis = balance_store.UsdCostBasisToFloat(prefetchedBalanceMetadata.usdCostBasis) } var mintMetadata *currencypb.Mint var liveReserveState *currencypb.VerifiedLaunchpadCurrencyReserveState @@ -542,7 +541,7 @@ func (s *server) getProtoAccountInfo(ctx context.Context, records *common.Accoun AccountType: records.General.AccountType, Index: records.General.Index, BalanceSource: prefetchedBalanceMetadata.source, - Balance: prefetchedBalanceMetadata.value, + Balance: prefetchedBalanceMetadata.quarks, UsdCostBasis: usdCostBasis, ManagementState: managementState, BlockchainState: blockchainState, diff --git a/ocp/rpc/transaction/gift_card.go b/ocp/rpc/transaction/gift_card.go index 21d8d72..e2f1321 100644 --- a/ocp/rpc/transaction/gift_card.go +++ b/ocp/rpc/transaction/gift_card.go @@ -12,7 +12,6 @@ import ( transactionpb "github.com/code-payments/ocp-protobuf-api/generated/go/transaction/v1" "github.com/code-payments/ocp-server/grpc/client" - "github.com/code-payments/ocp-server/ocp/balance" "github.com/code-payments/ocp-server/ocp/common" "github.com/code-payments/ocp-server/ocp/data/account" "github.com/code-payments/ocp-server/ocp/data/action" @@ -77,12 +76,6 @@ func (s *transactionServer) VoidGiftCard(ctx context.Context, req *transactionpb }, nil } - globalBalanceLock, err := balance.GetOptimisticVersionLock(ctx, s.data, giftCardVault) - if err != nil { - log.With(zap.Error(err)).Warn("failure getting balance lock") - return nil, status.Error(codes.Internal, "") - } - localAccountLock := s.getLocalAccountLock(giftCardVault) localAccountLock.Lock() defer localAccountLock.Unlock() @@ -131,7 +124,7 @@ func (s *transactionServer) VoidGiftCard(ctx context.Context, req *transactionpb }, nil } - err = account_worker.InitiateProcessToAutoReturnGiftCard(ctx, s.data, giftCardVault, true, globalBalanceLock) + err = account_worker.InitiateProcessToAutoReturnGiftCard(ctx, s.data, giftCardVault, true) if err != nil { log.With(zap.Error(err)).Warn("failure scheduling auto-return action") return nil, status.Error(codes.Internal, "") diff --git a/ocp/rpc/transaction/intent.go b/ocp/rpc/transaction/intent.go index 49e1d33..b2d1e2f 100644 --- a/ocp/rpc/transaction/intent.go +++ b/ocp/rpc/transaction/intent.go @@ -292,24 +292,24 @@ func (s *transactionServer) SubmitIntent(streamer transactionpb.Transaction_Subm return nil } - // Lock any acccounts with fund movement that is not resistent to race conditions - // 1. Global DB layer lock to guarantee balance consistency in a mult-server environment - // 2. Local in memory lock to avoid over consumption of local resources (eg. - // nonces) when we're likely to encounter a race resulting in DB txn rollback - // (eg. mass attempt to claim gift card). - globalBalanceLocks, err := intentHandler.GetBalanceLocks(ctx, intentRecord, submitActionsReq.Metadata) + // Locally lock any acccounts with fund movement to avoid over consumption + // of local resources (eg. nonces) when we're likely to encounter a race + // resulting in DB txn rollback (eg. mass attempt to claim gift card). + // Cross-server race resistance comes from the balance ledger's predicates + // in the intent DB transaction. + accountsToLock, err := intentHandler.GetAccountsToLock(ctx, intentRecord, submitActionsReq.Metadata) if err != nil { - log.With(zap.Error(err)).Warn("failure getting accounts with balances to lock") + log.With(zap.Error(err)).Warn("failure getting accounts to lock") return handleSubmitIntentError(ctx, streamer, intentRecord, err) } localAccountLocks := make([]*sync.Mutex, 0) locallyLockedAccounts := make(map[string]any) - for _, globalBalanceLock := range globalBalanceLocks { - _, ok := locallyLockedAccounts[globalBalanceLock.Account.PublicKey().ToBase58()] + for _, accountToLock := range accountsToLock { + _, ok := locallyLockedAccounts[accountToLock.PublicKey().ToBase58()] if !ok { - localAccountLocks = append(localAccountLocks, s.getLocalAccountLock(globalBalanceLock.Account)) + localAccountLocks = append(localAccountLocks, s.getLocalAccountLock(accountToLock)) } - locallyLockedAccounts[globalBalanceLock.Account.PublicKey().ToBase58()] = true + locallyLockedAccounts[accountToLock.PublicKey().ToBase58()] = true } for _, localAccountLock := range localAccountLocks { localAccountLock.Lock() @@ -717,14 +717,6 @@ func (s *transactionServer) SubmitIntent(streamer transactionpb.Transaction_Subm return err } - for _, globalBalanceLock := range globalBalanceLocks { - err = globalBalanceLock.CommitFn(ctx, s.data) - if err != nil { - log.With(zap.Error(err)).Warn("failure commiting balance update") - return err - } - } - // Reflect the intent in the balance ledger. Applied last, so the // ledger row locks are held for as little of the transaction as // possible. diff --git a/ocp/rpc/transaction/intent_handler.go b/ocp/rpc/transaction/intent_handler.go index 9cbce69..4585d38 100644 --- a/ocp/rpc/transaction/intent_handler.go +++ b/ocp/rpc/transaction/intent_handler.go @@ -34,15 +34,6 @@ import ( "github.com/code-payments/ocp-server/solana" ) -type intentBalanceLock struct { - // The account that's being locked - Account *common.Account - - // The function executed on intent DB commit that is guaranteed to prevent - // race conditions against invalid balance updates - CommitFn func(ctx context.Context, data ocp_data.Provider) error -} - // CreateIntentHandler is an interface for handling new intent creations type CreateIntentHandler interface { // PopulateMetadata adds intent metadata to the provided intent record @@ -64,9 +55,13 @@ type CreateIntentHandler interface { // error. IsNoop(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata, actions []*transactionpb.Action) (bool, error) - // GetBalanceLocks gets a set of global balance locks to prevent race conditions - // against invalid balance updates that would result in intent fulfillment failure - GetBalanceLocks(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*intentBalanceLock, error) + // GetAccountsToLock gets the set of accounts to lock in-process before + // validation, to avoid over consumption of local resources (eg. nonces) + // on requests that are likely to lose a race and roll back. Only expected + // mass races are worth locking, like a gift card being claimed from many + // devices at once. Race resistance itself comes from the balance ledger's + // predicates in the intent transaction. + GetAccountsToLock(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*common.Account, error) // AllowCreation determines whether the new intent creation should be allowed. AllowCreation(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata, actions []*transactionpb.Action) error @@ -192,7 +187,7 @@ func (h *OpenAccountsIntentHandler) IsNoop(ctx context.Context, intentRecord *in return accountInfoRecord.TokenAccount == tokenAccountToCheck.PublicKey().ToBase58() && accountInfoRecord.AccountType == expectedAccountType, nil } -func (h *OpenAccountsIntentHandler) GetBalanceLocks(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*intentBalanceLock, error) { +func (h *OpenAccountsIntentHandler) GetAccountsToLock(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*common.Account, error) { return nil, nil } @@ -508,58 +503,8 @@ func (h *SendPublicPaymentIntentHandler) IsNoop(ctx context.Context, intentRecor return false, nil } -func (h *SendPublicPaymentIntentHandler) GetBalanceLocks(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*intentBalanceLock, error) { - typedMetadata := metadata.GetSendPublicPayment() - if typedMetadata == nil { - return nil, errors.New("unexpected metadata proto message") - } - - sourceVault, err := common.NewAccountFromProto(typedMetadata.Source) - if err != nil { - return nil, err - } - - outgoingSourceBalanceLock, err := balance.GetOptimisticVersionLock(ctx, h.data, sourceVault) - if err != nil { - return nil, err - } - - intentBalanceLocks := []*intentBalanceLock{ - { - Account: sourceVault, - CommitFn: outgoingSourceBalanceLock.OnNewBalanceVersion, - }, - } - - if h.cachedDestinationAccountInfoRecord != nil { - switch h.cachedDestinationAccountInfoRecord.AccountType { - case commonpb.AccountType_POOL: - closeableDestinationVault, err := common.NewAccountFromProto(typedMetadata.Destination) - if err != nil { - return nil, err - } - - incomingDestinationBalanceLock, err := balance.GetOptimisticVersionLock(ctx, h.data, closeableDestinationVault) - if err != nil { - return nil, err - } - incomingDestinationOpenCloseLock := balance.NewOpenCloseStatusLock(closeableDestinationVault) - - intentBalanceLocks = append( - intentBalanceLocks, - &intentBalanceLock{ - Account: closeableDestinationVault, - CommitFn: incomingDestinationBalanceLock.RequireSameBalanceVerion, - }, - &intentBalanceLock{ - Account: closeableDestinationVault, - CommitFn: incomingDestinationOpenCloseLock.OnPaymentToAccount, - }, - ) - } - } - - return intentBalanceLocks, nil +func (h *SendPublicPaymentIntentHandler) GetAccountsToLock(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*common.Account, error) { + return nil, nil } func (h *SendPublicPaymentIntentHandler) AllowCreation(ctx context.Context, intentRecord *intent.Record, untypedMetadata *transactionpb.Metadata, actions []*transactionpb.Action) error { @@ -1088,23 +1033,16 @@ func (h *ReceivePaymentsPubliclyIntentHandler) IsNoop(ctx context.Context, inten return false, nil } -func (h *ReceivePaymentsPubliclyIntentHandler) GetBalanceLocks(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*intentBalanceLock, error) { +func (h *ReceivePaymentsPubliclyIntentHandler) GetAccountsToLock(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*common.Account, error) { + // A gift card claim is an expected mass race across devices, so requests + // on this server queue behind one lock instead of burning nonces on + // attempts that are doomed to lose the gift card's drain giftCardVault, err := common.NewAccountFromPublicKeyString(intentRecord.ReceivePaymentsPubliclyMetadata.Source) if err != nil { return nil, err } - outgoingGiftCardBalanceLock, err := balance.GetOptimisticVersionLock(ctx, h.data, giftCardVault) - if err != nil { - return nil, err - } - - return []*intentBalanceLock{ - { - Account: giftCardVault, - CommitFn: outgoingGiftCardBalanceLock.OnNewBalanceVersion, - }, - }, nil + return []*common.Account{giftCardVault}, nil } func (h *ReceivePaymentsPubliclyIntentHandler) AllowCreation(ctx context.Context, intentRecord *intent.Record, untypedMetadata *transactionpb.Metadata, actions []*transactionpb.Action) error { @@ -1437,29 +1375,8 @@ func (h *PublicDistributionIntentHandler) IsNoop(ctx context.Context, intentReco return false, nil } -func (h *PublicDistributionIntentHandler) GetBalanceLocks(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*intentBalanceLock, error) { - poolVault, err := common.NewAccountFromPublicKeyString(intentRecord.PublicDistributionMetadata.Source) - if err != nil { - return nil, err - } - - outgoingPoolBalanceLock, err := balance.GetOptimisticVersionLock(ctx, h.data, poolVault) - if err != nil { - return nil, err - } - - incomingPoolBalanceLock := balance.NewOpenCloseStatusLock(poolVault) - - return []*intentBalanceLock{ - { - Account: poolVault, - CommitFn: outgoingPoolBalanceLock.OnNewBalanceVersion, - }, - { - Account: poolVault, - CommitFn: incomingPoolBalanceLock.OnClose, - }, - }, nil +func (h *PublicDistributionIntentHandler) GetAccountsToLock(ctx context.Context, intentRecord *intent.Record, metadata *transactionpb.Metadata) ([]*common.Account, error) { + return nil, nil } // todo: Not all multi-mint validation checks are implemented diff --git a/ocp/worker/account/gift_card.go b/ocp/worker/account/gift_card.go index ac95600..031c15b 100644 --- a/ocp/worker/account/gift_card.go +++ b/ocp/worker/account/gift_card.go @@ -93,12 +93,6 @@ func (p *runtime) maybeInitiateGiftCardAutoReturn(ctx context.Context, accountIn return err } - balanceLock, err := balance.GetOptimisticVersionLock(ctx, p.data, giftCardVaultAccount) - if err != nil { - log.With(zap.Error(err)).Warn("failure getting balance lock") - return err - } - _, err = p.data.GetGiftCardClaimedAction(ctx, giftCardVaultAccount.PublicKey().ToBase58()) if err == nil { log.Debug("gift card is claimed and will be removed from worker queue") @@ -150,7 +144,7 @@ func (p *runtime) maybeInitiateGiftCardAutoReturn(ctx context.Context, accountIn // There's no action to claim the gift card and the expiry window has been met. // It's time to initiate the process of auto-returning the funds back to the // issuer. - err = InitiateProcessToAutoReturnGiftCard(ctx, p.data, giftCardVaultAccount, false, balanceLock) + err = InitiateProcessToAutoReturnGiftCard(ctx, p.data, giftCardVaultAccount, false) if err != nil { log.With(zap.Error(err)).Warn("failure initiating process to return gift card balance to issuer") return err @@ -169,7 +163,7 @@ func (p *runtime) maybeInitiateGiftCardAutoReturn(ctx context.Context, accountIn // a good guide for similar actions in the future. // // todo: This probably belongs somewhere more common -func InitiateProcessToAutoReturnGiftCard(ctx context.Context, data ocp_data.Provider, giftCardVaultAccount *common.Account, isVoidedByUser bool, balanceLock *balance.OptimisticVersionLock) error { +func InitiateProcessToAutoReturnGiftCard(ctx context.Context, data ocp_data.Provider, giftCardVaultAccount *common.Account, isVoidedByUser bool) error { return data.ExecuteInTx(ctx, sql.LevelDefault, func(ctx context.Context) error { giftCardIssuedIntent, err := data.GetOriginalGiftCardIssuedIntent(ctx, giftCardVaultAccount.PublicKey().ToBase58()) if err != nil { @@ -252,12 +246,7 @@ func InitiateProcessToAutoReturnGiftCard(ctx context.Context, data ocp_data.Prov // This will trigger the fulfillment worker to poll for the fulfillment. This // should be the very last DB update called. - err = markFulfillmentAsActivelyScheduled(ctx, data, autoReturnFulfillment[0]) - if err != nil { - return err - } - - return balanceLock.OnNewBalanceVersion(ctx, data) + return markFulfillmentAsActivelyScheduled(ctx, data, autoReturnFulfillment[0]) }) } diff --git a/ocp/worker/currency/holder/runtime.go b/ocp/worker/currency/holder/runtime.go index bf6f84b..72d78e4 100644 --- a/ocp/worker/currency/holder/runtime.go +++ b/ocp/worker/currency/holder/runtime.go @@ -115,6 +115,10 @@ func (p *holderRuntime) countHoldersForMint(ctx context.Context, mint string, cu return 0, nil } + if balance.LedgerReadsEnabled(ctx) { + return p.data.CountLockedBalancesByMint(ctx, mint, int64(minHoldings)) + } + accountRecords, err := p.data.GetAccountInfosByMintAndType(ctx, mint, commonpb.AccountType_PRIMARY) if err == account.ErrAccountInfoNotFound { return 0, nil