Rewind: Re-enable Rewinding (#1645)

* Revert "Disable rewinding and reject the  query param on account lookups and searches. (#1630)"

This reverts commit 249016c. The synthetic transaction implementation of payouts in the indexer should allow balances retrieved via rewind to calculate as before.

* GCI lint warning fix.

* Add HeartbeatTxn to be ignored during rewinding.

accounting/rewind.go

@@ -0,0 +1,189 @@
+package accounting
+
+import (
+	"context"
+	"fmt"
+
+	models "github.com/algorand/indexer/v3/api/generated/v2"
+	"github.com/algorand/indexer/v3/idb"
+	"github.com/algorand/indexer/v3/types"
+
+	sdk "github.com/algorand/go-algorand-sdk/v2/types"
+)
+
+// ConsistencyError is returned when the database returns inconsistent (stale) results.
+type ConsistencyError struct {
+	msg string
+}
+
+func (e ConsistencyError) Error() string {
+	return e.msg
+}
+
+func assetUpdate(account *models.Account, assetid uint64, add, sub uint64) {
+	if account.Assets == nil {
+		account.Assets = new([]models.AssetHolding)
+	}
+	assets := *account.Assets
+	for i, ah := range assets {
+		if ah.AssetId == assetid {
+			ah.Amount += add
+			ah.Amount -= sub
+			assets[i] = ah
+			// found and updated asset, done
+			return
+		}
+	}
+	// add asset to list
+	assets = append(assets, models.AssetHolding{
+		Amount:  add - sub,
+		AssetId: assetid,
+		//Creator: base32 addr string of asset creator, TODO
+		//IsFrozen: leave nil? // TODO: on close record frozen state for rewind
+	})
+	*account.Assets = assets
+}
+
+// SpecialAccountRewindError indicates that an attempt was made to rewind one of the special accounts.
+type SpecialAccountRewindError struct {
+	account string
+}
+
+// MakeSpecialAccountRewindError helper to initialize a SpecialAccountRewindError.
+func MakeSpecialAccountRewindError(account string) *SpecialAccountRewindError {
+	return &SpecialAccountRewindError{account: account}
+}
+
+// Error is part of the error interface.
+func (sare *SpecialAccountRewindError) Error() string {
+	return fmt.Sprintf("unable to rewind the %s", sare.account)
+}
+
+var specialAccounts *types.SpecialAddresses
+
+// AccountAtRound queries the idb.IndexerDb object for transactions and rewinds most fields of the account back to
+// their values at the requested round.
+// `round` must be <= `account.Round`
+func AccountAtRound(ctx context.Context, account models.Account, round uint64, db idb.IndexerDb) (acct models.Account, err error) {
+	// Make sure special accounts cache has been initialized.
+	if specialAccounts == nil {
+		var accounts types.SpecialAddresses
+		accounts, err = db.GetSpecialAccounts(ctx)
+		if err != nil {
+			return models.Account{}, fmt.Errorf("unable to get special accounts: %v", err)
+		}
+		specialAccounts = &accounts
+	}
+
+	acct = account
+	var addr sdk.Address
+	addr, err = sdk.DecodeAddress(account.Address)
+	if err != nil {
+		return
+	}
+
+	// ensure that the don't attempt to rewind a special account.
+	if specialAccounts.FeeSink == addr {
+		err = MakeSpecialAccountRewindError("FeeSink")
+		return
+	}
+	if specialAccounts.RewardsPool == addr {
+		err = MakeSpecialAccountRewindError("RewardsPool")
+		return
+	}
+
+	// Get transactions and rewind account.
+	tf := idb.TransactionFilter{
+		Address:  addr[:],
+		MinRound: round + 1,
+		MaxRound: account.Round,
+	}
+	ctx2, cf := context.WithCancel(ctx)
+	// In case of a panic before the next defer, call cf() here.
+	defer cf()
+	txns, r := db.Transactions(ctx2, tf)
+	// In case of an error, make sure the context is cancelled, and the channel is cleaned up.
+	defer func() {
+		cf()
+		for range txns {
+		}
+	}()
+	if r < account.Round {
+		err = ConsistencyError{fmt.Sprintf("queried round r: %d < account.Round: %d", r, account.Round)}
+		return
+	}
+	txcount := 0
+	for txnrow := range txns {
+		if txnrow.Error != nil {
+			err = txnrow.Error
+			return
+		}
+		txcount++
+		stxn := txnrow.Txn
+		if stxn == nil {
+			return models.Account{},
+				fmt.Errorf("rewinding past inner transactions is not supported")
+		}
+		if addr == stxn.Txn.Sender {
+			acct.AmountWithoutPendingRewards += uint64(stxn.Txn.Fee)
+			acct.AmountWithoutPendingRewards -= uint64(stxn.SenderRewards)
+		}
+		switch stxn.Txn.Type {
+		case sdk.PaymentTx:
+			if addr == stxn.Txn.Sender {
+				acct.AmountWithoutPendingRewards += uint64(stxn.Txn.Amount)
+			}
+			if addr == stxn.Txn.Receiver {
+				acct.AmountWithoutPendingRewards -= uint64(stxn.Txn.Amount)
+				acct.AmountWithoutPendingRewards -= uint64(stxn.ReceiverRewards)
+			}
+			if addr == stxn.Txn.CloseRemainderTo {
+				// unwind receiving a close-to
+				acct.AmountWithoutPendingRewards -= uint64(stxn.ClosingAmount)
+				acct.AmountWithoutPendingRewards -= uint64(stxn.CloseRewards)
+			} else if !stxn.Txn.CloseRemainderTo.IsZero() {
+				// unwind sending a close-to
+				acct.AmountWithoutPendingRewards += uint64(stxn.ClosingAmount)
+			}
+		case sdk.KeyRegistrationTx:
+			// TODO: keyreg does not rewind. workaround: query for txns on an account with typeenum=2 to find previous values it was set to.
+		case sdk.AssetConfigTx:
+			if stxn.Txn.ConfigAsset == 0 {
+				// create asset, unwind the application of the value
+				assetUpdate(&acct, txnrow.AssetID, 0, stxn.Txn.AssetParams.Total)
+			}
+		case sdk.AssetTransferTx:
+			if addr == stxn.Txn.AssetSender || addr == stxn.Txn.Sender {
+				assetUpdate(&acct, uint64(stxn.Txn.XferAsset), stxn.Txn.AssetAmount+txnrow.Extra.AssetCloseAmount, 0)
+			}
+			if addr == stxn.Txn.AssetReceiver {
+				assetUpdate(&acct, uint64(stxn.Txn.XferAsset), 0, stxn.Txn.AssetAmount)
+			}
+			if addr == stxn.Txn.AssetCloseTo {
+				assetUpdate(&acct, uint64(stxn.Txn.XferAsset), 0, txnrow.Extra.AssetCloseAmount)
+			}
+		case sdk.AssetFreezeTx:
+		case sdk.HeartbeatTx:
+		default:
+			err = fmt.Errorf("%s[%d,%d]: rewinding past txn type %s is not currently supported", account.Address, txnrow.Round, txnrow.Intra, stxn.Txn.Type)
+			return
+		}
+	}
+
+	acct.Round = round
+
+	// Due to accounts being closed and re-opened, we cannot always rewind Rewards. So clear it out.
+	acct.Rewards = 0
+
+	// Computing pending rewards is not supported.
+	acct.PendingRewards = 0
+	acct.Amount = acct.AmountWithoutPendingRewards
+
+	// MinBalance is not supported.
+	acct.MinBalance = 0
+
+	// TODO: Clear out the closed-at field as well. Like Rewards we cannot know this value for all accounts.
+	//acct.ClosedAt = 0
+
+	return
+}

accounting/rewind_test.go

@@ -0,0 +1,80 @@
+package accounting
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	models "github.com/algorand/indexer/v3/api/generated/v2"
+	"github.com/algorand/indexer/v3/idb"
+	"github.com/algorand/indexer/v3/idb/mocks"
+	"github.com/algorand/indexer/v3/types"
+
+	sdk "github.com/algorand/go-algorand-sdk/v2/types"
+)
+
+func TestBasic(t *testing.T) {
+	var a sdk.Address
+	a[0] = 'a'
+
+	account := models.Account{
+		Address:                     a.String(),
+		Amount:                      100,
+		AmountWithoutPendingRewards: 100,
+		Round:                       8,
+	}
+
+	txnRow := idb.TxnRow{
+		Round: 7,
+		Txn: &sdk.SignedTxnWithAD{
+			SignedTxn: sdk.SignedTxn{
+				Txn: sdk.Transaction{
+					Type: sdk.PaymentTx,
+					PaymentTxnFields: sdk.PaymentTxnFields{
+						Receiver: a,
+						Amount:   sdk.MicroAlgos(2),
+					},
+				},
+			},
+		},
+	}
+
+	ch := make(chan idb.TxnRow, 1)
+	ch <- txnRow
+	close(ch)
+	var outCh <-chan idb.TxnRow = ch
+
+	db := &mocks.IndexerDb{}
+	db.On("GetSpecialAccounts", mock.Anything).Return(types.SpecialAddresses{}, nil)
+	db.On("Transactions", mock.Anything, mock.Anything).Return(outCh, uint64(8))
+
+	account, err := AccountAtRound(context.Background(), account, 6, db)
+	assert.NoError(t, err)
+
+	assert.Equal(t, uint64(98), account.Amount)
+}
+
+// Test that when idb.Transactions() returns stale data the first time, we return an error.
+func TestStaleTransactions1(t *testing.T) {
+	var a sdk.Address
+	a[0] = 'a'
+
+	account := models.Account{
+		Address: a.String(),
+		Round:   8,
+	}
+
+	ch := make(chan idb.TxnRow)
+	var outCh <-chan idb.TxnRow = ch
+	close(ch)
+
+	db := &mocks.IndexerDb{}
+	db.On("GetSpecialAccounts", mock.Anything).Return(types.SpecialAddresses{}, nil)
+	db.On("Transactions", mock.Anything, mock.Anything).Return(outCh, uint64(7)).Once()
+
+	account, err := AccountAtRound(context.Background(), account, 6, db)
+	assert.True(t, errors.As(err, &ConsistencyError{}), "err: %v", err)
+}

api/error_messages.go

@@ -37,7 +37,8 @@ const (
 	errMultipleApplications            = "multiple applications found for this id, please contact us, this shouldn't happen"
 	ErrMultipleBoxes                   = "multiple application boxes found for this app id and box name, please contact us, this shouldn't happen"
 	ErrFailedLookingUpBoxes            = "failed while looking up application boxes"
-	errRewindingAccountNotSupported    = "rewinding account is no longer supported, please remove the `round=` query parameter and try again"
+	errMultiAcctRewind                 = "multiple accounts rewind is not supported by this server"
+	errRewindingAccount                = "error while rewinding account"
 	errLookingUpBlockForRound          = "error while looking up block for round"
 	errBlockHeaderSearch               = "error while searching for block headers"
 	errTransactionSearch               = "error while searching for transaction"