Skip to content
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

exp/lighthorizon/index: Allow accounts to be indexed by ledger. #4495

Merged
merged 5 commits into from
Aug 2, 2022
Merged

exp/lighthorizon/index: Allow accounts to be indexed by ledger. #4495

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6e8d236
Add builders to make account indices by ledger
Shaptic Aug 1, 2022
e0f7850
Add module parameter to map job in batch builder
Shaptic Aug 1, 2022
f17c4c2
Don't build transaction indices by default
Shaptic Aug 1, 2022
02867f8
Build both indices in the tests
Shaptic Aug 1, 2022
51741d1
Use the right variable :facepalm:
Shaptic Aug 2, 2022
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
9 changes: 7 additions & 2 deletions exp/lighthorizon/index/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,9 +88,14 @@ func BuildIndices(
case "transactions":
indexBuilder.RegisterModule(ProcessTransaction)
case "accounts":
indexBuilder.RegisterModule(ProcessAccounts)
indexBuilder.RegisterModule(ProcessAccountsByCheckpoint)
case "accounts_by_ledger":
indexBuilder.RegisterModule(ProcessAccountsByLedger)
case "accounts_unbacked":
indexBuilder.RegisterModule(ProcessAccountsWithoutBackend)
indexBuilder.RegisterModule(ProcessAccountsByCheckpointWithoutBackend)
indexStore.ClearMemory(false)
case "accounts_by_ledger_unbacked":
indexBuilder.RegisterModule(ProcessAccountsByLedgerWithoutBackend)
indexStore.ClearMemory(false)
default:
return indexBuilder, fmt.Errorf("unknown module '%s'", part)
Expand Down
53 changes: 29 additions & 24 deletions exp/lighthorizon/index/cmd/batch/map/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"os"
"runtime"
"strconv"
"strings"

"github.com/stellar/go/exp/lighthorizon/index"
"github.com/stellar/go/historyarchive"
Expand All @@ -29,6 +30,7 @@ const (
indexTargetUrlEnv = "INDEX_TARGET"
workerCountEnv = "WORKER_COUNT"
networkPassphraseEnv = "NETWORK_PASSPHRASE"
modulesEnv = "MODULES"
)

func NewBatchConfig() (*BatchConfig, error) {
Expand Down Expand Up @@ -68,28 +70,12 @@ func NewBatchConfig() (*BatchConfig, error) {
return nil, errors.New("required parameter " + txmetaSourceUrlEnv)
}

networkPassphrase := os.Getenv(networkPassphraseEnv)
switch networkPassphrase {
case "":
log.Warnf("%s not specified, defaulting to 'testnet'", networkPassphraseEnv)
fallthrough
case "testnet":
networkPassphrase = network.TestNetworkPassphrase
case "pubnet":
networkPassphrase = network.PublicNetworkPassphrase
default:
log.Warnf("%s is not a recognized shortcut ('pubnet' or 'testnet')",
networkPassphraseEnv)
}
log.Infof("Using network passphrase '%s'", networkPassphrase)

firstLedger := uint32(firstCheckpoint + batchSize*jobIndex)
lastLedger := firstLedger + uint32(batchSize) - 1
return &BatchConfig{
Range: historyarchive.Range{Low: firstLedger, High: lastLedger},
TxMetaSourceUrl: txmetaSourceUrl,
IndexTargetUrl: fmt.Sprintf("%s%cjob_%d", indexTargetRootUrl, os.PathSeparator, jobIndex),
NetworkPassphrase: networkPassphrase,
Range: historyarchive.Range{Low: firstLedger, High: lastLedger},
TxMetaSourceUrl: txmetaSourceUrl,
IndexTargetUrl: fmt.Sprintf("%s%cjob_%d", indexTargetRootUrl, os.PathSeparator, jobIndex),
}, nil
}

Expand All @@ -115,19 +101,38 @@ func main() {
workerCount = int(workerCountParsed)
}

networkPassphrase := os.Getenv(networkPassphraseEnv)
switch networkPassphrase {
case "":
log.Warnf("%s not specified, defaulting to 'testnet'", networkPassphraseEnv)
fallthrough
case "testnet":
networkPassphrase = network.TestNetworkPassphrase
case "pubnet":
networkPassphrase = network.PublicNetworkPassphrase
default:
log.Warnf("%s is not a recognized shortcut ('pubnet' or 'testnet')",
networkPassphraseEnv)
}
log.Infof("Using network passphrase '%s'", networkPassphrase)

parsedModules := []string{}
if modules := os.Getenv(modulesEnv); modules == "" {
parsedModules = append(parsedModules, "accounts_unbacked")
} else {
parsedModules = append(parsedModules, strings.Split(modules, ",")...)
}

log.Infof("Uploading ledger range [%d, %d] to %s",
batch.Range.Low, batch.Range.High, batch.IndexTargetUrl)

if _, err := index.BuildIndices(
context.Background(),
batch.TxMetaSourceUrl,
batch.IndexTargetUrl,
batch.NetworkPassphrase,
networkPassphrase,
batch.Range,
[]string{
"accounts_unbacked",
"transactions",
},
parsedModules,
workerCount,
); err != nil {
panic(err)
Expand Down
4 changes: 2 additions & 2 deletions exp/lighthorizon/index/cmd/map.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ for (( i=0; i < $BATCH_COUNT; i++ ))
do
echo -n "Creating map job $i... "

NETWORK_PASSPHRASE='testnet' \
NETWORK_PASSPHRASE='testnet' MODULES='accounts_unbacked,transactions' \
AWS_BATCH_JOB_ARRAY_INDEX=$i BATCH_SIZE=$BATCH_SIZE FIRST_CHECKPOINT=$FIRST \
TXMETA_SOURCE=file://$1 INDEX_TARGET=file://$2 WORKER_COUNT=1 \
./map &
Expand All @@ -93,4 +93,4 @@ done

rm ./map
echo "All jobs succeeded!"
exit 0
exit 0
101 changes: 68 additions & 33 deletions exp/lighthorizon/index/modules.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,13 @@ var (
checkpointManager = historyarchive.NewCheckpointManager(0)
)

const (
ByCheckpoint = iota
ByLedger = iota
)

type AccountIndexMode int

func ProcessTransaction(
indexStore Store,
ledger xdr.LedgerCloseMeta,
Expand All @@ -24,25 +31,39 @@ func ProcessTransaction(
)
}

// GetCheckpointNumber returns the next checkpoint NUMBER (NOT the checkpoint
// ledger sequence) corresponding to a given ledger sequence.
func GetCheckpointNumber(ledger uint32) uint32 {
return 1 + (ledger / checkpointManager.GetCheckpointFrequency())
func ProcessAccountsByCheckpoint(
indexStore Store,
ledger xdr.LedgerCloseMeta,
tx ingest.LedgerTransaction,
) error {
return ProcessAccounts(indexStore, ledger, tx, ByCheckpoint)
}

func ProcessAccountsByLedger(
indexStore Store,
ledger xdr.LedgerCloseMeta,
tx ingest.LedgerTransaction,
) error {
return ProcessAccounts(indexStore, ledger, tx, ByLedger)
}

func ProcessAccounts(
indexStore Store,
ledger xdr.LedgerCloseMeta,
tx ingest.LedgerTransaction,
mode AccountIndexMode,
) error {
checkpoint := GetCheckpointNumber(ledger.LedgerSequence())
index := getIndex(ledger, mode)
if index == 0 {
return fmt.Errorf("Invalid account indexing mode: %d", mode)
}

allParticipants, err := GetTransactionParticipants(tx)
if err != nil {
return err
}

err = indexStore.AddParticipantsToIndexes(checkpoint, "all/all", allParticipants)
err = indexStore.AddParticipantsToIndexes(index, "all/all", allParticipants)
if err != nil {
return err
}
Expand All @@ -52,39 +73,48 @@ func ProcessAccounts(
return err
}

err = indexStore.AddParticipantsToIndexes(checkpoint, "all/payments", paymentsParticipants)
err = indexStore.AddParticipantsToIndexes(index, "all/payments", paymentsParticipants)
if err != nil {
return err
}

// if tx.Result.Successful() {
// err = indexStore.AddParticipantsToIndexes(checkpoint, "successful/all", allParticipants)
// if err != nil {
// return err
// }
return nil
}

func ProcessAccountsByCheckpointWithoutBackend(
indexStore Store,
ledger xdr.LedgerCloseMeta,
tx ingest.LedgerTransaction,
) error {
return ProcessAccountsWithoutBackend(indexStore, ledger, tx, ByCheckpoint)
}

// err = indexStore.AddParticipantsToIndexes(checkpoint, "successful/payments", paymentsParticipants)
// if err != nil {
// return err
// }
// }
func ProcessAccountsByLedgerWithoutBackend(
indexStore Store,
ledger xdr.LedgerCloseMeta,
tx ingest.LedgerTransaction,
) error {
return ProcessAccountsWithoutBackend(indexStore, ledger, tx, ByLedger)

return nil
}

func ProcessAccountsWithoutBackend(
indexStore Store,
ledger xdr.LedgerCloseMeta,
tx ingest.LedgerTransaction,
mode AccountIndexMode,
) error {
checkpoint := GetCheckpointNumber(ledger.LedgerSequence())
index := getIndex(ledger, mode)
if index == 0 {
return fmt.Errorf("Invalid account indexing mode: %d", mode)
}

allParticipants, err := GetTransactionParticipants(tx)
if err != nil {
return err
}

err = indexStore.AddParticipantsToIndexesNoBackend(checkpoint, "all/all", allParticipants)
err = indexStore.AddParticipantsToIndexesNoBackend(index, "all/all", allParticipants)
if err != nil {
return err
}
Expand All @@ -94,26 +124,20 @@ func ProcessAccountsWithoutBackend(
return err
}

err = indexStore.AddParticipantsToIndexesNoBackend(checkpoint, "all/payments", paymentsParticipants)
err = indexStore.AddParticipantsToIndexesNoBackend(index, "all/payments", paymentsParticipants)
if err != nil {
return err
}

// if tx.Result.Successful() {
// err = indexStore.AddParticipantsToIndexesNoBackend(checkpoint, "successful/all", allParticipants)
// if err != nil {
// return err
// }

// err = indexStore.AddParticipantsToIndexesNoBackend(checkpoint, "successful/payments", paymentsParticipants)
// if err != nil {
// return err
// }
// }

return nil
}

// GetCheckpointNumber returns the next checkpoint NUMBER (NOT the checkpoint
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you think there's value in having this in historyarchive.CheckpointManager, inlined here is good as well because it avoids dependency into mono repo, we're already using CheckpointManager elsewhere in lighthorizon.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting idea! I'm not entirely sure -- this detail is specific to the way indexes are built. We might change the way it's represented in the future, e.g. what if we decide a different way to "bucket" them? I guess that's a really opaque notion... I'm for moving it as it stands today. I'll open a separate PR!

// ledger sequence) corresponding to a given ledger sequence.
func GetCheckpointNumber(ledger uint32) uint32 {
return 1 + (ledger / checkpointManager.GetCheckpointFrequency())
}

func GetPaymentParticipants(transaction ingest.LedgerTransaction) ([]string, error) {
return participantsForOperations(transaction, true)
}
Expand Down Expand Up @@ -277,3 +301,14 @@ func getLedgerKeyParticipants(ledgerKey xdr.LedgerKey) []string {
}
return []string{}
}

func getIndex(ledger xdr.LedgerCloseMeta, mode AccountIndexMode) uint32 {
switch mode {
case ByCheckpoint:
return GetCheckpointNumber(ledger.LedgerSequence())
case ByLedger:
return ledger.LedgerSequence()
default:
return 0
}
}