-
Notifications
You must be signed in to change notification settings - Fork 504
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
exp/lighthorizon: Isolate cursor advancement code to its own interface (
#4484) * Move cursor manipulation code to a separate interface * Small test refactor to improve readability and long-running lines * Combine tx and op tests into subtests * Fix how IndexStore is mocked out
- Loading branch information
Showing
5 changed files
with
379 additions
and
191 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,98 @@ | ||
| package services | ||
|
|
||
| import ( | ||
| "github.com/stellar/go/exp/lighthorizon/index" | ||
| "github.com/stellar/go/toid" | ||
| ) | ||
|
|
||
| // CursorManager describes a way to control how a cursor advances for a | ||
| // particular indexing strategy. | ||
| type CursorManager interface { | ||
| Begin(cursor int64) (int64, error) | ||
| Advance() (int64, error) | ||
| } | ||
|
|
||
| type AccountActivityCursorManager struct { | ||
| AccountId string | ||
|
|
||
| store index.Store | ||
| lastCursor *toid.ID | ||
| } | ||
|
|
||
| func NewCursorManagerForAccountActivity(store index.Store, accountId string) *AccountActivityCursorManager { | ||
| return &AccountActivityCursorManager{AccountId: accountId, store: store} | ||
| } | ||
|
|
||
| func (c *AccountActivityCursorManager) Begin(cursor int64) (int64, error) { | ||
| freq := checkpointManager.GetCheckpointFrequency() | ||
| id := toid.Parse(cursor) | ||
| lastCheckpoint := uint32(0) | ||
| if id.LedgerSequence >= int32(checkpointManager.GetCheckpointFrequency()) { | ||
| lastCheckpoint = index.GetCheckpointNumber(uint32(id.LedgerSequence)) | ||
| } | ||
|
|
||
| // We shouldn't take the provided cursor for granted: instead, we should | ||
| // skip ahead to the first active ledger that's >= the given cursor. | ||
| // | ||
| // For example, someone might say ?cursor=0 but the first active checkpoint | ||
| // is actually 40M ledgers in. | ||
| firstCheckpoint, err := c.store.NextActive(c.AccountId, allTransactionsIndex, lastCheckpoint) | ||
| if err != nil { | ||
| return cursor, err | ||
| } | ||
|
|
||
| nextLedger := (firstCheckpoint - 1) * freq | ||
|
|
||
| // However, if the given cursor is actually *more* specific than the index | ||
| // can give us (e.g. somewhere *within* an active checkpoint range), prefer | ||
| // it rather than starting over. | ||
| if nextLedger < uint32(id.LedgerSequence) { | ||
| better := toid.Parse(cursor) | ||
| c.lastCursor = &better | ||
| return cursor, nil | ||
| } | ||
|
|
||
| c.lastCursor = toid.New(int32(nextLedger), 1, 1) | ||
| return c.lastCursor.ToInt64(), nil | ||
| } | ||
|
|
||
| func (c *AccountActivityCursorManager) Advance() (int64, error) { | ||
| if c.lastCursor == nil { | ||
| panic("invalid cursor, call Begin() first") | ||
| } | ||
|
|
||
| // Advancing the cursor means deciding whether or not we need to query the | ||
| // index. | ||
|
|
||
| lastLedger := uint32(c.lastCursor.LedgerSequence) | ||
| freq := checkpointManager.GetCheckpointFrequency() | ||
|
|
||
| if checkpointManager.IsCheckpoint(lastLedger) { | ||
| // If the last cursor we looked at was a checkpoint ledger, then we need | ||
| // to jump ahead to the next checkpoint. Note that NextActive() is | ||
| // "inclusive" so if the parameter is an active checkpoint it will | ||
| // return itself. | ||
| checkpoint := index.GetCheckpointNumber(uint32(c.lastCursor.LedgerSequence)) | ||
| checkpoint, err := c.store.NextActive(c.AccountId, allTransactionsIndex, checkpoint+1) | ||
| if err != nil { | ||
| return c.lastCursor.ToInt64(), err | ||
| } | ||
|
|
||
| // We add a -1 here because an active checkpoint indicates that an | ||
| // account had activity in the *previous* 64 ledgers, so we need to | ||
| // backtrack to that ledger range. | ||
| c.lastCursor = toid.New(int32((checkpoint-1)*freq), 1, 1) | ||
| } else { | ||
| // Otherwise, we can just bump the ledger number. | ||
| c.lastCursor = toid.New(int32(lastLedger+1), 1, 1) | ||
| } | ||
|
|
||
| return c.lastCursor.ToInt64(), nil | ||
| } | ||
|
|
||
| var _ CursorManager = (*AccountActivityCursorManager)(nil) // ensure conformity to the interface | ||
|
|
||
| // getLedgerFromCursor is a helpful way to turn a cursor into a ledger number | ||
| func getLedgerFromCursor(cursor int64) uint32 { | ||
| return uint32(toid.Parse(cursor).LedgerSequence) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| package services | ||
|
|
||
| import ( | ||
| "io" | ||
| "testing" | ||
|
|
||
| "github.com/stellar/go/exp/lighthorizon/index" | ||
| "github.com/stellar/go/historyarchive" | ||
| "github.com/stellar/go/keypair" | ||
| "github.com/stellar/go/toid" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| var ( | ||
| checkpointMgr = historyarchive.NewCheckpointManager(0) | ||
| ) | ||
|
|
||
| func TestAccountTransactionCursorManager(t *testing.T) { | ||
| freq := int32(checkpointMgr.GetCheckpointFrequency()) | ||
| accountId := keypair.MustRandom().Address() | ||
|
|
||
| // Create an index and fill it with some checkpoint details. | ||
| store, err := index.NewFileStore(index.StoreConfig{ | ||
| Url: "file://" + t.TempDir(), | ||
| Workers: 4, | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| for _, checkpoint := range []uint32{1, 5, 10} { | ||
| require.NoError(t, store.AddParticipantsToIndexes( | ||
| checkpoint, allTransactionsIndex, []string{accountId})) | ||
| } | ||
|
|
||
| cursorMgr := NewCursorManagerForAccountActivity(store, accountId) | ||
|
|
||
| cursor := toid.New(1, 1, 1) | ||
| var nextCursor int64 | ||
|
|
||
| // first checkpoint works | ||
| nextCursor, err = cursorMgr.Begin(cursor.ToInt64()) | ||
| require.NoError(t, err) | ||
| assert.EqualValues(t, 1, getLedgerFromCursor(nextCursor)) | ||
|
|
||
| // cursor is preserved if mid-active-range | ||
| cursor.LedgerSequence = freq / 2 | ||
| nextCursor, err = cursorMgr.Begin(cursor.ToInt64()) | ||
| require.NoError(t, err) | ||
| assert.EqualValues(t, cursor.LedgerSequence, getLedgerFromCursor(nextCursor)) | ||
|
|
||
| // cursor jumps ahead if not active | ||
| cursor.LedgerSequence = 2 * freq | ||
| nextCursor, err = cursorMgr.Begin(cursor.ToInt64()) | ||
| require.NoError(t, err) | ||
| assert.EqualValues(t, 4*freq, getLedgerFromCursor(nextCursor)) | ||
|
|
||
| for i := int32(1); i < freq; i++ { | ||
| nextCursor, err = cursorMgr.Advance() | ||
| require.NoError(t, err) | ||
| assert.EqualValues(t, 4*freq+i, getLedgerFromCursor(nextCursor)) | ||
| } | ||
|
|
||
| // cursor jumps to next active checkpoint | ||
| nextCursor, err = cursorMgr.Advance() | ||
| require.NoError(t, err) | ||
| assert.EqualValues(t, 9*freq, getLedgerFromCursor(nextCursor)) | ||
|
|
||
| // cursor increments | ||
| for i := int32(1); i < freq; i++ { | ||
| nextCursor, err = cursorMgr.Advance() | ||
| require.NoError(t, err) | ||
| assert.EqualValues(t, 9*freq+i, getLedgerFromCursor(nextCursor)) | ||
| } | ||
|
|
||
| // cursor stops when no more actives | ||
| _, err = cursorMgr.Advance() | ||
| assert.ErrorIs(t, err, io.EOF) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.