Skip to content

Fix reclaim-rent sweeps and surface failures #1045

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
raymondjacobson merged 1 commit into from
Sep 24, 2026
Merged

Fix reclaim-rent sweeps and surface failures #1045

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 7 additions & 6 deletions cmd/reclaim_rent/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ func init() {
reclaimRentCmd.Flags().StringP("rpc", "r", "https://api.mainnet-beta.solana.com", "The Solana RPC endpoint to use")
reclaimRentCmd.Flags().StringP("database", "c", "postgres://postgres:postgres@localhost:5432/discovery_provider_1?sslmode=disable", "Database connection string")
reclaimRentCmd.Flags().StringP("keypair", "k", "~/.config/solana/id.json", "The wallet to use as fee payer for transactions")
reclaimRentCmd.Flags().StringP("destination", "d", "", "The recipient of reclaimed rent (defaults to fee payer)")
reclaimRentCmd.Flags().StringP("destination", "d", claimable_tokens.DefaultRentDestinationAddress, "The recipient of reclaimed rent (must match the claimable-tokens program)")
reclaimRentCmd.Flags().StringP("program", "p", claimable_tokens.ProgramID.String(), "The claimable tokens program ID")
reclaimRentCmd.Flags().StringP("created-after", "", "", "Filter accounts created after this date (RFC3339 format, e.g., 2024-01-01T00:00:00Z)")
}
Expand Down Expand Up @@ -77,12 +77,13 @@ func reclaimRent(cmd *cobra.Command, args []string) error {
if err != nil {
return fmt.Errorf("failed to get destination flag: %w", err)
}
var destination solana.PublicKey
if destinationFlag == "" {
destination = keypair.PublicKey()
} else {
destination = solana.MustPublicKeyFromBase58(destinationFlag)
if destinationFlag != claimable_tokens.DefaultRentDestinationAddress {
return fmt.Errorf(
"destination must be %s; the claimable-tokens program rejects other addresses",
claimable_tokens.DefaultRentDestinationAddress,
)
}
destination := solana.MustPublicKeyFromBase58(destinationFlag)

programIDFlag, err := cmd.Flags().GetString("program")
if err != nil {
Expand Down
186 changes: 170 additions & 16 deletions jobs/reclaim_rent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package jobs

import (
"context"
"errors"
"fmt"
"sync"
"time"
Expand Down Expand Up @@ -30,25 +31,43 @@ const (
type ReclaimRentJob struct {
cfg config.Config
pool database.DbPool
rpcClient *rpc.Client
transactionSender *spl.TransactionSender
rpcClient reclaimRentRPCClient
transactionSender reclaimRentTransactionSender
mints []solana.PublicKey
logger *zap.Logger

mutex sync.Mutex
isRunning bool
}

type reclaimRentRPCClient interface {
GetMultipleAccountsWithOpts(
context.Context,
[]solana.PublicKey,
*rpc.GetMultipleAccountsOpts,
) (*rpc.GetMultipleAccountsResult, error)
}

type reclaimRentTransactionSender interface {
GetFeePayer() (*solana.Wallet, error)
SendTransactionWithRetries(
context.Context,
*solana.TransactionBuilder,
rpc.CommitmentType,
rpc.TransactionOpts,
) (*solana.Signature, error)
}

func NewReclaimRentJob(cfg config.Config, pool database.DbPool) *ReclaimRentJob {
logger := logging.NewZapLogger(cfg).Named("ReclaimRentJob")

var rpcClient *rpc.Client
if len(cfg.SolanaConfig.RpcProviders) > 0 {
var rpcClient reclaimRentRPCClient
if len(cfg.SolanaConfig.RpcProviders) > 0 && cfg.SolanaConfig.RpcProviders[0] != "" {
rpcClient = rpc.New(cfg.SolanaConfig.RpcProviders[0])
}

var transactionSender *spl.TransactionSender
if len(cfg.SolanaConfig.RpcProviders) > 0 {
var transactionSender reclaimRentTransactionSender
if rpcClient != nil {
transactionSender = spl.NewTransactionSender(cfg.SolanaConfig.FeePayers, cfg.SolanaConfig.RpcProviders)
}

Expand Down Expand Up @@ -102,7 +121,8 @@ func (j *ReclaimRentJob) Run(ctx context.Context) {

// Closes zero-balance claimable token accounts created in the last 7 days
// for the configured AUDIO and USDC mints, returning the rent lamports to the
// fee payer that signs each transaction. Ensures only one instance runs at a time.
// destination required by the claimable-tokens program. Ensures only one
// instance runs at a time.
func (j *ReclaimRentJob) run(ctx context.Context) error {
j.mutex.Lock()
if j.isRunning {
Expand All @@ -126,6 +146,7 @@ func (j *ReclaimRentJob) run(ctx context.Context) error {
return nil
}

var runErrors []error
for _, mint := range j.mints {
if mint.IsZero() {
continue
Expand All @@ -135,16 +156,39 @@ func (j *ReclaimRentJob) run(ctx context.Context) error {
zap.String("mint", mint.String()),
zap.Error(err),
)
runErrors = append(runErrors, fmt.Errorf("process mint %s: %w", mint, err))
}
}
return nil
return errors.Join(runErrors...)
}

type reclaimRentAccount struct {
Account string `db:"account"`
EthereumAddress string `db:"ethereum_address"`
}

type reclaimRentFilterStats struct {
MissingAccount int
InvalidOwner int
InvalidData int
MintMismatch int
NonZeroBalance int
InvalidEthAddress int
AddressMismatch int
WrongCloseAuthority int
}

func (s *reclaimRentFilterStats) add(other reclaimRentFilterStats) {
s.MissingAccount += other.MissingAccount
s.InvalidOwner += other.InvalidOwner
s.InvalidData += other.InvalidData
s.MintMismatch += other.MintMismatch
s.NonZeroBalance += other.NonZeroBalance
s.InvalidEthAddress += other.InvalidEthAddress
s.AddressMismatch += other.AddressMismatch
s.WrongCloseAuthority += other.WrongCloseAuthority
}

func (j *ReclaimRentJob) processMint(ctx context.Context, mint solana.PublicKey) error {
logger := j.logger.With(zap.String("mint", mint.String()))
logger.Info("Processing mint")
Expand All @@ -156,7 +200,19 @@ func (j *ReclaimRentJob) processMint(ctx context.Context, mint solana.PublicKey)

cutoff := time.Now().Add(-reclaimRentLookback)
offset := 0
totalCandidates := 0
totalClosable := 0
totalClosed := 0
totalFailed := 0
var filterStats reclaimRentFilterStats
var firstProcessError error
processErrorCount := 0
recordProcessError := func(err error) {
processErrorCount++
if firstProcessError == nil {
firstProcessError = err
}
}
for {
accounts, err := j.fetchCandidates(ctx, mint.String(), cutoff, reclaimRentDbPageSize, offset)
if err != nil {
Expand All @@ -166,12 +222,16 @@ func (j *ReclaimRentJob) processMint(ctx context.Context, mint solana.PublicKey)
break
}
offset += len(accounts)
totalCandidates += len(accounts)

filtered, err := j.filterOnChain(ctx, accounts)
filtered, pageFilterStats, err := j.filterOnChain(ctx, accounts, mint, authority)
if err != nil {
logger.Error("filterOnChain failed", zap.Error(err))
recordProcessError(err)
continue
}
filterStats.add(pageFilterStats)
totalClosable += len(filtered)

for i := 0; i < len(filtered); i += reclaimRentBatchSize {
end := i + reclaimRentBatchSize
Expand All @@ -185,6 +245,18 @@ func (j *ReclaimRentJob) processMint(ctx context.Context, mint solana.PublicKey)
zap.Error(err),
zap.Int("batch_size", len(batch)),
)
totalFailed += len(batch)
recordProcessError(err)
continue
}
if sig == nil {
err := errors.New("transaction sender returned a nil signature")
logger.Error("processBatch failed",
zap.Error(err),
zap.Int("batch_size", len(batch)),
)
totalFailed += len(batch)
recordProcessError(err)
continue
}
logger.Info("Reclaimed batch",
Expand All @@ -194,7 +266,29 @@ func (j *ReclaimRentJob) processMint(ctx context.Context, mint solana.PublicKey)
totalClosed += len(batch)
}
}
logger.Info("Done processing mint", zap.Int("total_closed", totalClosed))
logger.Info("Done processing mint",
zap.Int("db_candidates", totalCandidates),
zap.Int("onchain_closable", totalClosable),
zap.Int("total_closed", totalClosed),
zap.Int("total_failed", totalFailed),
zap.Int("processing_errors", processErrorCount),
zap.Int("skipped_missing", filterStats.MissingAccount),
zap.Int("skipped_invalid_owner", filterStats.InvalidOwner),
zap.Int("skipped_invalid_data", filterStats.InvalidData),
zap.Int("skipped_mint_mismatch", filterStats.MintMismatch),
zap.Int("skipped_nonzero_balance", filterStats.NonZeroBalance),
zap.Int("skipped_invalid_eth_address", filterStats.InvalidEthAddress),
zap.Int("skipped_address_mismatch", filterStats.AddressMismatch),
zap.Int("skipped_wrong_close_authority", filterStats.WrongCloseAuthority),
)
if firstProcessError != nil {
return fmt.Errorf(
"%d processing operation(s) failed (%d close attempts affected): %w",
processErrorCount,
totalFailed,
firstProcessError,
)
}
return nil
}

Expand Down Expand Up @@ -222,33 +316,85 @@ func (j *ReclaimRentJob) fetchCandidates(ctx context.Context, mint string, since
return pgx.CollectRows(rows, pgx.RowToStructByName[reclaimRentAccount])
}

func (j *ReclaimRentJob) filterOnChain(ctx context.Context, batch []reclaimRentAccount) ([]reclaimRentAccount, error) {
func (j *ReclaimRentJob) filterOnChain(
ctx context.Context,
batch []reclaimRentAccount,
mint solana.PublicKey,
authority solana.PublicKey,
) ([]reclaimRentAccount, reclaimRentFilterStats, error) {
var stats reclaimRentFilterStats
pubkeys := make([]solana.PublicKey, 0, len(batch))
for _, acct := range batch {
pubkeys = append(pubkeys, solana.MustPublicKeyFromBase58(acct.Account))
pubkey, err := solana.PublicKeyFromBase58(acct.Account)
if err != nil {
return nil, stats, fmt.Errorf("invalid account public key %q: %w", acct.Account, err)
}
pubkeys = append(pubkeys, pubkey)
}
res, err := j.rpcClient.GetMultipleAccountsWithOpts(ctx, pubkeys, &rpc.GetMultipleAccountsOpts{
Encoding: solana.EncodingBase64,
})
if err != nil {
return nil, fmt.Errorf("failed to get accounts: %w", err)
return nil, stats, fmt.Errorf("failed to get accounts: %w", err)
}
if len(res.Value) != len(batch) {
return nil, stats, fmt.Errorf("RPC returned %d accounts for a batch of %d", len(res.Value), len(batch))
}

filtered := make([]reclaimRentAccount, 0, len(batch))
for i, info := range res.Value {
if info == nil {
stats.MissingAccount++
continue
}
if !info.Owner.Equals(solana.TokenProgramID) {
stats.InvalidOwner++
continue
}
if info.Data == nil {
stats.InvalidData++
continue
}
var ta token.Account
if err := bin.NewBorshDecoder(info.Data.GetBinary()).Decode(&ta); err != nil {
stats.InvalidData++
continue
}
if !ta.Mint.Equals(mint) {
stats.MintMismatch++
continue
}
if ta.Amount != 0 {
stats.NonZeroBalance++
continue
}
if !common.IsHexAddress(batch[i].EthereumAddress) {
stats.InvalidEthAddress++
continue
}
expectedUserBank, err := claimable_tokens.DeriveUserBankAccount(
mint,
common.HexToAddress(batch[i].EthereumAddress),
)
if err != nil {
return nil, stats, fmt.Errorf("derive user bank for %q: %w", batch[i].Account, err)
}
if !pubkeys[i].Equals(expectedUserBank) {
stats.AddressMismatch++
continue
}

closeAuthority := ta.Owner
if ta.CloseAuthority != nil {
closeAuthority = *ta.CloseAuthority
}
if !closeAuthority.Equals(authority) {
stats.WrongCloseAuthority++
continue
}
filtered = append(filtered, batch[i])
}
return filtered, nil
return filtered, stats, nil
}

func (j *ReclaimRentJob) processBatch(ctx context.Context, batch []reclaimRentAccount, authority solana.PublicKey) (*solana.Signature, error) {
Expand All @@ -262,11 +408,19 @@ func (j *ReclaimRentJob) processBatch(ctx context.Context, batch []reclaimRentAc
}

builder := solana.NewTransactionBuilder().SetFeePayer(payer.PublicKey())
rentDestination := solana.MustPublicKeyFromBase58(claimable_tokens.DefaultRentDestinationAddress)
for _, acct := range batch {
userBank, err := solana.PublicKeyFromBase58(acct.Account)
if err != nil {
return nil, fmt.Errorf("invalid account public key %q: %w", acct.Account, err)
}
if !common.IsHexAddress(acct.EthereumAddress) {
return nil, fmt.Errorf("invalid Ethereum address %q for account %s", acct.EthereumAddress, acct.Account)
}
inst := claimable_tokens.NewCloseInstructionBuilder().
SetUserBank(solana.MustPublicKeyFromBase58(acct.Account)).
SetUserBank(userBank).
SetAuthority(authority).
SetDestination(payer.PublicKey()).
SetDestination(rentDestination).
SetEthAddress(common.HexToAddress(acct.EthereumAddress))
builder.AddInstruction(inst.Build())
}
Expand Down
Loading
Loading