diff --git a/ocp/worker/currency/feeburner/config.go b/ocp/worker/currency/feeburner/config.go new file mode 100644 index 0000000..a1af67c --- /dev/null +++ b/ocp/worker/currency/feeburner/config.go @@ -0,0 +1,34 @@ +package feeburner + +import ( + "github.com/code-payments/ocp-server/config" + "github.com/code-payments/ocp-server/config/env" +) + +const ( + envConfigPrefix = "CURRENCY_FEE_BURNER_RUNTIME_" + + SubsidizerConfigEnvName = envConfigPrefix + "SUBSIDIZER" + defaultSubsidizer = "invalid" + + BatchSizeConfigEnvName = envConfigPrefix + "WORKER_BATCH_SIZE" + defaultBatchSize = 100 +) + +type conf struct { + subsidizer config.String + batchSize config.Uint64 +} + +// ConfigProvider defines how config values are pulled +type ConfigProvider func() *conf + +// WithEnvConfigs returns configuration pulled from environment variables +func WithEnvConfigs() ConfigProvider { + return func() *conf { + return &conf{ + subsidizer: env.NewStringConfig(SubsidizerConfigEnvName, defaultSubsidizer), + batchSize: env.NewUint64Config(BatchSizeConfigEnvName, defaultBatchSize), + } + } +} diff --git a/ocp/worker/currency/feeburner/runtime.go b/ocp/worker/currency/feeburner/runtime.go new file mode 100644 index 0000000..a298e1c --- /dev/null +++ b/ocp/worker/currency/feeburner/runtime.go @@ -0,0 +1,70 @@ +package feeburner + +import ( + "context" + "time" + + "github.com/pkg/errors" + "go.uber.org/zap" + + "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/worker" +) + +type runtime struct { + log *zap.Logger + conf *conf + data ocp_data.Provider + subsidizer *common.Account +} + +func New(log *zap.Logger, data ocp_data.Provider, configProvider ConfigProvider) (worker.Runtime, error) { + p := &runtime{ + log: log, + conf: configProvider(), + data: data, + } + + err := p.loadSubsidizer() + if err != nil { + return nil, err + } + return p, nil +} + +func (p *runtime) Start(ctx context.Context, interval time.Duration) error { + // Fees are burned immediately on startup, then once per interval. The + // cadence is intentionally not durable: burning early (eg. after a + // restart) is safe because BurnFees only burns what has accumulated + // since the last burn. + for { + delay := interval + + err := p.sweep(ctx) + if err != nil && err != context.Canceled { + p.log.With(zap.Error(err)).Warn("failure sweeping currencies for fee burning") + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(delay): + } + } +} + +func (p *runtime) loadSubsidizer() error { + ctx := context.TODO() + + vaultRecord, err := p.data.GetKey(ctx, p.conf.subsidizer.Get(ctx)) + if err != nil { + return errors.Wrap(err, "error getting subsidizer vault record") + } + + p.subsidizer, err = common.NewAccountFromPrivateKeyString(vaultRecord.PrivateKey) + if err != nil { + return errors.Wrap(err, "invalid subsidizer private key") + } + return nil +} diff --git a/ocp/worker/currency/feeburner/worker.go b/ocp/worker/currency/feeburner/worker.go new file mode 100644 index 0000000..eff6238 --- /dev/null +++ b/ocp/worker/currency/feeburner/worker.go @@ -0,0 +1,162 @@ +package feeburner + +import ( + "context" + + "github.com/pkg/errors" + "go.uber.org/zap" + + "github.com/code-payments/ocp-server/database/query" + "github.com/code-payments/ocp-server/metrics" + "github.com/code-payments/ocp-server/ocp/common" + "github.com/code-payments/ocp-server/ocp/data/currency" + transaction_util "github.com/code-payments/ocp-server/ocp/transaction" + "github.com/code-payments/ocp-server/solana" + compute_budget "github.com/code-payments/ocp-server/solana/computebudget" + "github.com/code-payments/ocp-server/solana/currencycreator" +) + +const ( + burnFeesComputeUnitLimit = 200_000 + computeUnitPrice = 10_000 +) + +type burnTarget struct { + mint string + ixn solana.Instruction +} + +func (p *runtime) sweep(runtimeCtx context.Context) error { + provider := runtimeCtx.Value(metrics.ProviderContextKey).(metrics.Provider) + trace := provider.StartTrace("currency_fee_burner_runtime__sweep") + defer trace.End() + tracedCtx := metrics.NewContext(runtimeCtx, trace) + + var cursor query.Cursor + for { + items, err := p.data.GetAllCurrencyMetadataByState( + tracedCtx, + currency.MetadataStateAvailable, + query.WithLimit(p.conf.batchSize.Get(tracedCtx)), + query.WithCursor(cursor), + ) + if err == currency.ErrNotFound { + return nil + } else if err != nil { + trace.OnError(err) + return err + } + + targets := make([]*burnTarget, 0, len(items)) + for _, item := range items { + target, err := p.makeBurnTarget(item) + if err != nil { + trace.OnError(err) + p.log.With( + zap.Error(err), + zap.String("mint", item.Mint), + ).Warn("skipping currency with invalid metadata") + continue + } + targets = append(targets, target) + } + + for _, batch := range p.packBurnBatches(targets) { + err := p.burnFeesForBatch(tracedCtx, batch) + if err != nil { + trace.OnError(err) + p.log.With( + zap.Error(err), + zap.Int("batch_size", len(batch)), + ).Warn("failure burning fees for batch") + } + } + + cursor = query.ToCursor(items[len(items)-1].Id) + } +} + +func (p *runtime) makeBurnTarget(record *currency.MetadataRecord) (*burnTarget, error) { + poolAccount, err := common.NewAccountFromPublicKeyString(record.LiquidityPool) + if err != nil { + return nil, errors.Wrap(err, "invalid liquidity pool") + } + + vaultCoreAccount, err := common.NewAccountFromPublicKeyString(record.VaultCore) + if err != nil { + return nil, errors.Wrap(err, "invalid core vault") + } + + return &burnTarget{ + mint: record.Mint, + ixn: currencycreator.NewBurnFeesInstruction( + ¤cycreator.BurnFeesInstructionAccounts{ + Payer: p.subsidizer.PublicKey().ToBytes(), + Pool: poolAccount.PublicKey().ToBytes(), + BaseMint: common.CoreMintAccount.PublicKey().ToBytes(), + VaultBase: vaultCoreAccount.PublicKey().ToBytes(), + }, + ¤cycreator.BurnFeesInstructionArgs{}, + ), + }, nil +} + +// packBurnBatches greedily packs burn targets into the fewest transactions +// that fit within the transaction size limit. +func (p *runtime) packBurnBatches(targets []*burnTarget) [][]*burnTarget { + var batches [][]*burnTarget + var current []*burnTarget + for _, target := range targets { + candidate := append(current, target) + txn := p.makeBurnTransaction(candidate) + if len(txn.Marshal()) > solana.MaxTransactionSize { + if len(current) == 0 { + p.log.With(zap.String("mint", target.mint)).Warn("skipping currency with oversized burn transaction") + continue + } + batches = append(batches, current) + current = []*burnTarget{target} + continue + } + current = candidate + } + if len(current) > 0 { + batches = append(batches, current) + } + return batches +} + +func (p *runtime) makeBurnTransaction(batch []*burnTarget) solana.Transaction { + ixns := make([]solana.Instruction, 0, len(batch)+2) + ixns = append( + ixns, + compute_budget.SetComputeUnitLimit(uint32(len(batch))*burnFeesComputeUnitLimit), + compute_budget.SetComputeUnitPrice(computeUnitPrice), + ) + for _, target := range batch { + ixns = append(ixns, target.ixn) + } + return solana.NewLegacyTransaction(p.subsidizer.PublicKey().ToBytes(), ixns...) +} + +func (p *runtime) burnFeesForBatch(ctx context.Context, batch []*burnTarget) error { + txn := p.makeBurnTransaction(batch) + + bh, err := p.data.GetBlockchainLatestBlockhash(ctx) + if err != nil { + return errors.Wrap(err, "error getting latest blockhash") + } + txn.SetBlockhash(bh) + + err = txn.Sign(p.subsidizer.PrivateKey().ToBytes()) + if err != nil { + return errors.Wrap(err, "error signing transaction") + } + + err = transaction_util.SubmitAndWaitForFinalization(ctx, p.data, &txn) + if err != nil { + return errors.Wrap(err, "error submitting transaction") + } + + return nil +} diff --git a/ocp/worker/currency/feeburner/worker_test.go b/ocp/worker/currency/feeburner/worker_test.go new file mode 100644 index 0000000..af0522f --- /dev/null +++ b/ocp/worker/currency/feeburner/worker_test.go @@ -0,0 +1,69 @@ +package feeburner + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + + "github.com/code-payments/ocp-server/ocp/data/currency" + "github.com/code-payments/ocp-server/solana" + "github.com/code-payments/ocp-server/testutil" +) + +func TestPackBurnBatches(t *testing.T) { + p := &runtime{ + log: zap.NewNop(), + subsidizer: testutil.NewRandomAccount(t), + } + + var targets []*burnTarget + for range 40 { + record := ¤cy.MetadataRecord{ + Mint: testutil.NewRandomAccount(t).PublicKey().ToBase58(), + LiquidityPool: testutil.NewRandomAccount(t).PublicKey().ToBase58(), + VaultCore: testutil.NewRandomAccount(t).PublicKey().ToBase58(), + } + + target, err := p.makeBurnTarget(record) + require.NoError(t, err) + targets = append(targets, target) + } + + batches := p.packBurnBatches(targets) + + var flattened []*burnTarget + for _, batch := range batches { + flattened = append(flattened, batch...) + } + require.Len(t, flattened, len(targets)) + for i, target := range targets { + assert.Equal(t, target.mint, flattened[i].mint) + } + + for i, batch := range batches { + txn := p.makeBurnTransaction(batch) + assert.LessOrEqual(t, len(txn.Marshal()), solana.MaxTransactionSize, fmt.Sprintf("batch %d exceeds size limit", i)) + } + + // Every batch except the last must be full: adding the next target would + // exceed the transaction size limit + for i := 0; i < len(batches)-1; i++ { + overfilled := append(append([]*burnTarget{}, batches[i]...), batches[i+1][0]) + txn := p.makeBurnTransaction(overfilled) + assert.Greater(t, len(txn.Marshal()), solana.MaxTransactionSize, fmt.Sprintf("batch %d is not fully packed", i)) + } + + assert.Greater(t, len(batches[0]), 1) +} + +func TestPackBurnBatches_Empty(t *testing.T) { + p := &runtime{ + log: zap.NewNop(), + subsidizer: testutil.NewRandomAccount(t), + } + + assert.Empty(t, p.packBurnBatches(nil)) +}