Add fee burn worker

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),
+		}
+	}
+}

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
+}

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(
+			&currencycreator.BurnFeesInstructionAccounts{
+				Payer:     p.subsidizer.PublicKey().ToBytes(),
+				Pool:      poolAccount.PublicKey().ToBytes(),
+				BaseMint:  common.CoreMintAccount.PublicKey().ToBytes(),
+				VaultBase: vaultCoreAccount.PublicKey().ToBytes(),
+			},
+			&currencycreator.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
+}

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 := &currency.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))
+}