diff --git a/cmd/gl-loadtest/README.md b/cmd/gl-loadtest/README.md new file mode 100644 index 0000000..371befb --- /dev/null +++ b/cmd/gl-loadtest/README.md @@ -0,0 +1,22 @@ +# GL load test + +This command exercises the write and read paths of the gRPC ledger service. +Each simulated transaction: + +1. appends a `CREATED` lifecycle event; +2. posts a balanced transfer between two generated user wallet accounts; +3. reads the committed journal; +4. rebuilds the receiving wallet balance; +5. appends a `SUCCESSFUL` lifecycle event. + +Ledger accounts are created implicitly by `AppendJournal`; GL does not expose a +separate wallet-creation RPC. + +Start GL, then run: + +```bash +go run ./cmd/gl-loadtest -duration 30s -concurrency 20 -wallets 1000 +``` + +Use a fresh `-run-id` for every dataset. The command generates one automatically +when the flag is omitted. diff --git a/cmd/gl-loadtest/main.go b/cmd/gl-loadtest/main.go new file mode 100644 index 0000000..b495625 --- /dev/null +++ b/cmd/gl-loadtest/main.go @@ -0,0 +1,323 @@ +// Command gl-loadtest runs a stateful gRPC load test against the GL service. +// Wallet accounts are created implicitly by the first journal that references +// each generated owner ID, matching normal GL behavior. +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "flag" + "fmt" + "os" + "os/signal" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + domain "gl/domain/ledger" + basev1 "gl/gen/base/v1" + ledgerv1 "gl/gen/ledger/v1" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/timestamppb" +) + +type client interface { + AppendJournal(context.Context, *ledgerv1.AppendJournalRequest, ...grpc.CallOption) (*ledgerv1.AppendJournalResponse, error) + AppendTransactionEvent(context.Context, *ledgerv1.AppendTransactionEventRequest, ...grpc.CallOption) (*ledgerv1.AppendTransactionEventResponse, error) + GetJournal(context.Context, *ledgerv1.GetJournalRequest, ...grpc.CallOption) (*ledgerv1.Journal, error) + GetBalance(context.Context, *ledgerv1.GetBalanceRequest, ...grpc.CallOption) (*ledgerv1.GetBalanceResponse, error) +} + +type config struct { + RunID string + Wallets int + AssetID int64 + Amount string + RequestTimeout time.Duration +} + +type sample struct { + Operation string + Latency time.Duration + Err error +} + +type operationStats struct { + Requests int64 + Errors int64 + Latency []time.Duration +} + +func main() { + target := flag.String("target", "127.0.0.1:8600", "GL gRPC target") + concurrency := flag.Int("concurrency", 20, "number of concurrent transaction workers") + duration := flag.Duration("duration", 30*time.Second, "load duration") + wallets := flag.Int("wallets", 1000, "number of simulated wallet accounts") + assetID := flag.Int64("asset-id", 1, "asset ID used by transfers") + amount := flag.String("amount", "1", "canonical decimal transfer amount") + requestTimeout := flag.Duration("request-timeout", 5*time.Second, "timeout for each gRPC call") + runID := flag.String("run-id", "", "idempotency namespace; generated when empty") + flag.Parse() + + if *concurrency < 1 || *duration <= 0 || *wallets < 2 || *assetID <= 0 || *requestTimeout <= 0 || *amount == "" { + fmt.Fprintln(os.Stderr, "concurrency and durations must be positive; wallets must be at least 2; asset-id and amount are required") + os.Exit(2) + } + parsedAmount, err := domain.ParseAmount(*amount) + if err != nil || parsedAmount.IsZero() || strings.HasPrefix(*amount, "-") { + fmt.Fprintf(os.Stderr, "amount must be a positive canonical ledger decimal: %v\n", err) + os.Exit(2) + } + if *runID == "" { + *runID = "gl-load-" + strconv.FormatInt(time.Now().UTC().UnixNano(), 36) + } + + connection, err := grpc.NewClient(*target, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + fmt.Fprintf(os.Stderr, "create gRPC client: %v\n", err) + os.Exit(2) + } + defer connection.Close() + ledger := ledgerv1.NewGeneralLedgerServiceClient(connection) + healthCtx, cancelHealth := context.WithTimeout(context.Background(), *requestTimeout) + health, err := ledger.Health(healthCtx, &basev1.Empty{}) + cancelHealth() + if err != nil || !health.GetServing() || !health.GetDatabaseReady() { + fmt.Fprintf(os.Stderr, "GL is not ready at %s: response=%v error=%v\n", *target, health, err) + os.Exit(1) + } + + testConfig := config{ + RunID: *runID, + Wallets: *wallets, + AssetID: *assetID, + Amount: *amount, + RequestTimeout: *requestTimeout, + } + signalCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + ctx, cancel := context.WithTimeout(signalCtx, *duration) + defer cancel() + + fmt.Printf("GL load test: target=%s workers=%d duration=%s wallets=%d asset=%d run=%s\n", *target, *concurrency, *duration, *wallets, *assetID, *runID) + started := time.Now() + samples, transactions := run(ctx, ledger, testConfig, *concurrency) + elapsed := time.Since(started) + failed := printSummary(samples, transactions, elapsed) + if failed { + os.Exit(1) + } +} + +func run(ctx context.Context, ledger client, cfg config, concurrency int) ([]sample, uint64) { + var sequence atomic.Uint64 + results := make(chan []sample, concurrency) + var workers sync.WaitGroup + workers.Add(concurrency) + for range concurrency { + go func() { + defer workers.Done() + workerSamples := make([]sample, 0, 256) + for ctx.Err() == nil { + transaction := sequence.Add(1) + workerSamples = append(workerSamples, runTransaction(ctx, ledger, cfg, transaction)...) + } + results <- workerSamples + }() + } + workers.Wait() + close(results) + + all := make([]sample, 0) + for workerSamples := range results { + all = append(all, workerSamples...) + } + return all, sequence.Load() +} + +func runTransaction(ctx context.Context, ledger client, cfg config, sequence uint64) []sample { + transactionID := fmt.Sprintf("%s-%d", cfg.RunID, sequence) + senderID := fmt.Sprintf("load-wallet-%d", sequence%uint64(cfg.Wallets)) + receiverID := fmt.Sprintf("load-wallet-%d", (sequence+1)%uint64(cfg.Wallets)) + correlationID := "load-correlation-" + transactionID + blockchainHash := transactionHash(transactionID) + occurredAt := timestamppb.Now() + + created := &ledgerv1.AppendTransactionEventRequest{ + SourceService: "gl-loadtest", + IdempotencyKey: "load:" + transactionID + ":event:v1", + SourceTransactionId: transactionID, + TrackingCode: "load-" + strconv.FormatUint(sequence, 10), + EventVersion: 1, + State: ledgerv1.TransactionState_TRANSACTION_STATE_CREATED, + OccurredAt: occurredAt, + CorrelationId: correlationID, + ActorId: senderID, + Metadata: map[string]string{"load_test": cfg.RunID}, + } + samples := []sample{measure(ctx, cfg.RequestTimeout, "append_event", func(callCtx context.Context) error { + _, err := ledger.AppendTransactionEvent(callCtx, created) + return err + })} + + journalRequest := &ledgerv1.AppendJournalRequest{ + SourceService: "gl-loadtest", + IdempotencyKey: "load:" + transactionID + ":transfer:v1", + SourceTransactionId: transactionID, + TrackingCode: created.TrackingCode, + EffectKind: "internal-transfer", + EventVersion: 1, + OccurredAt: occurredAt, + CorrelationId: correlationID, + ActorId: senderID, + Blockchain: &ledgerv1.BlockchainReference{ + Network: "load-test", + TransactionHash: blockchainHash, + LedgerSequence: strconv.FormatUint(sequence, 10), + }, + Metadata: map[string]string{"load_test": cfg.RunID}, + Entries: []*ledgerv1.JournalEntry{ + { + LineNumber: 1, + Account: walletAccount(senderID, cfg.AssetID), + Amount: "-" + cfg.Amount, + Description: "load-test transfer debit", + }, + { + LineNumber: 2, + Account: walletAccount(receiverID, cfg.AssetID), + Amount: cfg.Amount, + Description: "load-test transfer credit", + }, + }, + } + var journalID string + samples = append(samples, measure(ctx, cfg.RequestTimeout, "append_journal", func(callCtx context.Context) error { + response, err := ledger.AppendJournal(callCtx, journalRequest) + if err == nil && response.GetJournal() != nil { + journalID = response.GetJournal().GetJournalId() + } + return err + })) + + if journalID != "" { + samples = append(samples, measure(ctx, cfg.RequestTimeout, "get_journal", func(callCtx context.Context) error { + _, err := ledger.GetJournal(callCtx, &ledgerv1.GetJournalRequest{ + Lookup: &ledgerv1.GetJournalRequest_JournalId{JournalId: journalID}, + }) + return err + })) + } + + samples = append(samples, measure(ctx, cfg.RequestTimeout, "get_balance", func(callCtx context.Context) error { + _, err := ledger.GetBalance(callCtx, &ledgerv1.GetBalanceRequest{Account: walletAccount(receiverID, cfg.AssetID)}) + return err + })) + + successful := &ledgerv1.AppendTransactionEventRequest{ + SourceService: "gl-loadtest", + IdempotencyKey: "load:" + transactionID + ":event:v2", + SourceTransactionId: transactionID, + TrackingCode: created.TrackingCode, + EventVersion: 2, + State: ledgerv1.TransactionState_TRANSACTION_STATE_SUCCESSFUL, + OccurredAt: timestamppb.Now(), + CorrelationId: correlationID, + ActorId: senderID, + Blockchain: journalRequest.Blockchain, + Metadata: map[string]string{"load_test": cfg.RunID}, + } + samples = append(samples, measure(ctx, cfg.RequestTimeout, "append_event", func(callCtx context.Context) error { + _, err := ledger.AppendTransactionEvent(callCtx, successful) + return err + })) + return samples +} + +func walletAccount(ownerID string, assetID int64) *ledgerv1.AccountReference { + return &ledgerv1.AccountReference{ + AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE, + OwnerType: "user", + OwnerId: ownerID, + AssetId: assetID, + } +} + +func transactionHash(transactionID string) string { + value := sha256.Sum256([]byte(transactionID)) + return hex.EncodeToString(value[:]) +} + +func measure(ctx context.Context, timeout time.Duration, operation string, call func(context.Context) error) sample { + callCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + defer cancel() + started := time.Now() + err := call(callCtx) + return sample{Operation: operation, Latency: time.Since(started), Err: err} +} + +func printSummary(samples []sample, transactions uint64, elapsed time.Duration) bool { + stats := make(map[string]*operationStats) + var failed bool + for _, value := range samples { + operation := stats[value.Operation] + if operation == nil { + operation = &operationStats{} + stats[value.Operation] = operation + } + operation.Requests++ + operation.Latency = append(operation.Latency, value.Latency) + if value.Err != nil { + operation.Errors++ + failed = true + } + } + + names := make([]string, 0, len(stats)) + for name := range stats { + names = append(names, name) + } + sort.Strings(names) + fmt.Printf("transactions: %d (%.1f tx/s)\n", transactions, rate(float64(transactions), elapsed)) + fmt.Printf("%-18s %10s %8s %10s %12s %12s %12s\n", "operation", "requests", "errors", "req/s", "p50", "p95", "p99") + for _, name := range names { + value := stats[name] + fmt.Printf("%-18s %10d %8d %10.1f %12s %12s %12s\n", + name, + value.Requests, + value.Errors, + rate(float64(value.Requests), elapsed), + percentile(value.Latency, 50), + percentile(value.Latency, 95), + percentile(value.Latency, 99), + ) + } + return failed +} + +func rate(count float64, elapsed time.Duration) float64 { + if elapsed <= 0 { + return 0 + } + return count / elapsed.Seconds() +} + +func percentile(values []time.Duration, percent int) time.Duration { + if len(values) == 0 { + return 0 + } + ordered := append([]time.Duration(nil), values...) + sort.Slice(ordered, func(left, right int) bool { return ordered[left] < ordered[right] }) + index := (len(ordered)*percent + 99) / 100 + if index < 1 { + index = 1 + } + return ordered[index-1] +} diff --git a/cmd/gl-loadtest/main_test.go b/cmd/gl-loadtest/main_test.go new file mode 100644 index 0000000..faca7f5 --- /dev/null +++ b/cmd/gl-loadtest/main_test.go @@ -0,0 +1,79 @@ +package main + +import ( + "context" + "testing" + "time" + + ledgerv1 "gl/gen/ledger/v1" + + "google.golang.org/grpc" +) + +type clientStub struct { + journal *ledgerv1.AppendJournalRequest + events []*ledgerv1.AppendTransactionEventRequest + balance *ledgerv1.GetBalanceRequest + lookup *ledgerv1.GetJournalRequest +} + +func (s *clientStub) AppendJournal(_ context.Context, request *ledgerv1.AppendJournalRequest, _ ...grpc.CallOption) (*ledgerv1.AppendJournalResponse, error) { + s.journal = request + return &ledgerv1.AppendJournalResponse{Journal: &ledgerv1.Journal{JournalId: "journal-1"}}, nil +} + +func (s *clientStub) AppendTransactionEvent(_ context.Context, request *ledgerv1.AppendTransactionEventRequest, _ ...grpc.CallOption) (*ledgerv1.AppendTransactionEventResponse, error) { + s.events = append(s.events, request) + return &ledgerv1.AppendTransactionEventResponse{}, nil +} + +func (s *clientStub) GetJournal(_ context.Context, request *ledgerv1.GetJournalRequest, _ ...grpc.CallOption) (*ledgerv1.Journal, error) { + s.lookup = request + return &ledgerv1.Journal{JournalId: "journal-1"}, nil +} + +func (s *clientStub) GetBalance(_ context.Context, request *ledgerv1.GetBalanceRequest, _ ...grpc.CallOption) (*ledgerv1.GetBalanceResponse, error) { + s.balance = request + return &ledgerv1.GetBalanceResponse{Balance: "1"}, nil +} + +func TestRunTransactionCreatesWalletAccountsAndFullLifecycle(t *testing.T) { + ledger := &clientStub{} + samples := runTransaction(context.Background(), ledger, config{ + RunID: "test-run", Wallets: 10, AssetID: 7, Amount: "2.5", RequestTimeout: time.Second, + }, 3) + + if len(samples) != 5 { + t.Fatalf("expected five RPC measurements, got %d", len(samples)) + } + if ledger.journal == nil || len(ledger.journal.Entries) != 2 { + t.Fatal("balanced journal was not appended") + } + debit, credit := ledger.journal.Entries[0], ledger.journal.Entries[1] + if debit.Amount != "-2.5" || credit.Amount != "2.5" { + t.Fatalf("unexpected journal amounts: %q and %q", debit.Amount, credit.Amount) + } + if debit.Account.OwnerId == credit.Account.OwnerId || debit.Account.AssetId != 7 || credit.Account.AssetId != 7 { + t.Fatalf("unexpected simulated wallets: debit=%+v credit=%+v", debit.Account, credit.Account) + } + if len(ledger.events) != 2 || ledger.events[0].State != ledgerv1.TransactionState_TRANSACTION_STATE_CREATED || ledger.events[1].State != ledgerv1.TransactionState_TRANSACTION_STATE_SUCCESSFUL { + t.Fatalf("unexpected transaction lifecycle: %+v", ledger.events) + } + if ledger.lookup.GetJournalId() != "journal-1" || ledger.balance.Account.OwnerId != credit.Account.OwnerId { + t.Fatal("journal and balance reads did not follow the write") + } +} + +func TestTransactionHashIsStableAndUnique(t *testing.T) { + first := transactionHash("run-1") + if len(first) != 64 || first != transactionHash("run-1") || first == transactionHash("run-2") { + t.Fatalf("unexpected transaction hashes: %q", first) + } +} + +func TestPercentileUsesNearestRank(t *testing.T) { + values := []time.Duration{time.Millisecond, 4 * time.Millisecond, 2 * time.Millisecond, 3 * time.Millisecond} + if got := percentile(values, 95); got != 4*time.Millisecond { + t.Fatalf("unexpected p95: %s", got) + } +} diff --git a/cmd/gl-redistribution-loadtest/README.md b/cmd/gl-redistribution-loadtest/README.md new file mode 100644 index 0000000..d57bc6f --- /dev/null +++ b/cmd/gl-redistribution-loadtest/README.md @@ -0,0 +1,28 @@ +# GL redistribution load test + +This fixed-count scenario creates ten run-scoped user accounts for an operator- +selected asset. It verifies that all ten start at zero, funds only the first +person with 10,000, and then submits 10,000,000 user-to-user transfers. + +The first nine transfers give every other person 45. The remaining transfers +use reproducible random senders, receivers, and positive integer amounts while +reserving a minimum balance of 45 for every person. The test finishes by reading +all ten balances from GL and fails unless every balance is at least 45 and their +sum is exactly 10,000. + +Start GL, then run: + +```bash +go run ./cmd/gl-redistribution-loadtest +Asset number: 7 +``` + +For automation, pass the asset without the prompt: + +```bash +go run ./cmd/gl-redistribution-loadtest -asset-id 7 -concurrency 32 +``` + +The defaults are 10,000,000 transfers against `127.0.0.1:8600`. Use a fresh +`-run-id` for every dataset; one is generated automatically. `-transactions` +can be reduced for a smoke test, and `-seed` makes the random sequence repeatable. diff --git a/cmd/gl-redistribution-loadtest/main.go b/cmd/gl-redistribution-loadtest/main.go new file mode 100644 index 0000000..9738f90 --- /dev/null +++ b/cmd/gl-redistribution-loadtest/main.go @@ -0,0 +1,458 @@ +// Command gl-redistribution-loadtest proves balance conservation while applying +// a fixed number of random transfers between ten isolated test users. +package main + +import ( + "context" + "flag" + "fmt" + "io" + mathrand "math/rand" + "os" + "os/signal" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + basev1 "gl/gen/base/v1" + ledgerv1 "gl/gen/ledger/v1" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/timestamppb" +) + +const ( + personCount = 10 + initialSupply = int64(10_000) + minimumBalance = int64(45) + defaultTransactions = uint64(10_000_000) + seedTransactions = uint64(personCount - 1) +) + +type client interface { + Health(context.Context, *basev1.Empty, ...grpc.CallOption) (*ledgerv1.HealthResponse, error) + AppendJournal(context.Context, *ledgerv1.AppendJournalRequest, ...grpc.CallOption) (*ledgerv1.AppendJournalResponse, error) + GetBalance(context.Context, *ledgerv1.GetBalanceRequest, ...grpc.CallOption) (*ledgerv1.GetBalanceResponse, error) +} + +type config struct { + RunID string + AssetID int64 + Transactions uint64 + Concurrency int + RequestTimeout time.Duration + RandomSeed int64 +} + +type transfer struct { + Sender int + Receiver int + Amount int64 +} + +type balanceState struct { + mu sync.Mutex + balances [personCount]int64 +} + +type metrics struct { + attempted atomic.Uint64 + succeeded atomic.Uint64 + errors atomic.Uint64 + latencyNanos atomic.Int64 + maxNanos atomic.Int64 +} + +func main() { + os.Exit(runCLI(os.Args[1:], os.Stdin, os.Stdout, os.Stderr)) +} + +func runCLI(args []string, stdin io.Reader, stdout, stderr io.Writer) int { + flags := flag.NewFlagSet("gl-redistribution-loadtest", flag.ContinueOnError) + flags.SetOutput(stderr) + target := flags.String("target", "127.0.0.1:8600", "GL gRPC target") + assetID := flags.Int64("asset-id", 0, "asset number; prompted when omitted") + transactions := flags.Uint64("transactions", defaultTransactions, "total user-to-user transfers") + concurrency := flags.Int("concurrency", 32, "number of concurrent transfer workers") + requestTimeout := flags.Duration("request-timeout", 10*time.Second, "timeout for each gRPC call") + runID := flags.String("run-id", "", "isolated idempotency namespace; generated when empty") + randomSeed := flags.Int64("seed", 0, "random seed; generated when zero") + if err := flags.Parse(args); err != nil { + return 2 + } + if *transactions < seedTransactions || *concurrency < 1 || *requestTimeout <= 0 { + fmt.Fprintf(stderr, "transactions must be at least %d; concurrency and request-timeout must be positive\n", seedTransactions) + return 2 + } + + resolvedAssetID, err := resolveAssetID(*assetID, stdin, stdout) + if err != nil { + fmt.Fprintf(stderr, "asset number: %v\n", err) + return 2 + } + if *runID == "" { + *runID = "gl-redistribution-" + strconv.FormatInt(time.Now().UTC().UnixNano(), 36) + } + if *randomSeed == 0 { + *randomSeed = time.Now().UnixNano() + } + + connection, err := grpc.NewClient(*target, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + fmt.Fprintf(stderr, "create gRPC client: %v\n", err) + return 2 + } + defer connection.Close() + ledger := ledgerv1.NewGeneralLedgerServiceClient(connection) + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + cfg := config{ + RunID: *runID, + AssetID: resolvedAssetID, + Transactions: *transactions, + Concurrency: *concurrency, + RequestTimeout: *requestTimeout, + RandomSeed: *randomSeed, + } + if err := execute(ctx, ledger, cfg, stdout); err != nil { + fmt.Fprintf(stderr, "redistribution load test failed: %v\n", err) + return 1 + } + return 0 +} + +func resolveAssetID(value int64, input io.Reader, output io.Writer) (int64, error) { + if value > 0 { + return value, nil + } + if value < 0 { + return 0, fmt.Errorf("must be positive") + } + fmt.Fprint(output, "Asset number: ") + var text string + if _, err := fmt.Fscan(input, &text); err != nil { + return 0, fmt.Errorf("read input: %w", err) + } + assetID, err := strconv.ParseInt(strings.TrimSpace(text), 10, 64) + if err != nil || assetID <= 0 { + return 0, fmt.Errorf("must be a positive integer") + } + return assetID, nil +} + +func execute(ctx context.Context, ledger client, cfg config, output io.Writer) error { + callCtx, cancel := context.WithTimeout(ctx, cfg.RequestTimeout) + health, err := ledger.Health(callCtx, &basev1.Empty{}) + cancel() + if err != nil { + return fmt.Errorf("health check: %w", err) + } + if !health.GetServing() || !health.GetDatabaseReady() { + return fmt.Errorf("GL is not ready: %v", health) + } + + people := personIDs(cfg.RunID) + startingBalances, err := readBalances(ctx, ledger, cfg, people) + if err != nil { + return fmt.Errorf("read unfunded balances: %w", err) + } + if err := verifyUnfunded(startingBalances); err != nil { + return err + } + if err := fundFirstPerson(ctx, ledger, cfg, people[0]); err != nil { + return fmt.Errorf("fund first person: %w", err) + } + initialBalances, err := readBalances(ctx, ledger, cfg, people) + if err != nil { + return fmt.Errorf("read initial balances: %w", err) + } + if err := verifyInitial(initialBalances); err != nil { + return err + } + + fmt.Fprintf(output, "GL redistribution load test: target users=%d asset=%d transfers=%d workers=%d seed=%d run=%s\n", + personCount, cfg.AssetID, cfg.Transactions, cfg.Concurrency, cfg.RandomSeed, cfg.RunID) + fmt.Fprintln(output, "initial balances: [10000 0 0 0 0 0 0 0 0 0]") + + stats := &metrics{} + started := time.Now() + balances := initialBalances + for person := 1; person < personCount; person++ { + value := transfer{Sender: 0, Receiver: person, Amount: minimumBalance} + startedCall := time.Now() + err := appendTransfer(ctx, ledger, cfg, uint64(person), people, value) + stats.observe(time.Since(startedCall), err) + if err != nil { + return fmt.Errorf("seed person %d: %w", person+1, err) + } + balances[0] -= minimumBalance + balances[person] += minimumBalance + } + + state := &balanceState{balances: balances} + if err := runRandomTransfers(ctx, ledger, cfg, people, state, stats, output); err != nil { + printSummary(output, stats, time.Since(started)) + return err + } + elapsed := time.Since(started) + printSummary(output, stats, elapsed) + + finalBalances, err := readBalances(ctx, ledger, cfg, people) + if err != nil { + return fmt.Errorf("read final balances: %w", err) + } + if err := verifyFinal(finalBalances); err != nil { + return err + } + fmt.Fprintf(output, "final balances: %v\n", finalBalances) + fmt.Fprintf(output, "PASS: all %d people have at least %d; total=%d\n", personCount, minimumBalance, initialSupply) + return nil +} + +func personIDs(runID string) [personCount]string { + var people [personCount]string + for index := range people { + people[index] = fmt.Sprintf("%s-person-%02d", runID, index+1) + } + return people +} + +func readBalances(ctx context.Context, ledger client, cfg config, people [personCount]string) ([personCount]int64, error) { + var balances [personCount]int64 + for index, person := range people { + callCtx, cancel := context.WithTimeout(ctx, cfg.RequestTimeout) + response, err := ledger.GetBalance(callCtx, &ledgerv1.GetBalanceRequest{Account: personAccount(person, cfg.AssetID)}) + cancel() + if err != nil { + return balances, fmt.Errorf("person %d: %w", index+1, err) + } + if response == nil { + return balances, fmt.Errorf("person %d: empty response", index+1) + } + balance, err := strconv.ParseInt(response.GetBalance(), 10, 64) + if err != nil { + return balances, fmt.Errorf("person %d: invalid integer balance %q", index+1, response.GetBalance()) + } + balances[index] = balance + } + return balances, nil +} + +func verifyUnfunded(balances [personCount]int64) error { + for index, balance := range balances { + if balance != 0 { + return fmt.Errorf("person %d must start unfunded; got %d (use a fresh run-id)", index+1, balance) + } + } + return nil +} + +func verifyInitial(balances [personCount]int64) error { + if balances[0] != initialSupply { + return fmt.Errorf("person 1 initial balance: got %d, want %d", balances[0], initialSupply) + } + for index := 1; index < personCount; index++ { + if balances[index] != 0 { + return fmt.Errorf("person %d initial balance: got %d, want 0", index+1, balances[index]) + } + } + return nil +} + +func verifyFinal(balances [personCount]int64) error { + var total int64 + for index, balance := range balances { + if balance < minimumBalance { + return fmt.Errorf("person %d final balance %d is below %d", index+1, balance, minimumBalance) + } + total += balance + } + if total != initialSupply { + return fmt.Errorf("final balance sum: got %d, want %d", total, initialSupply) + } + return nil +} + +func fundFirstPerson(ctx context.Context, ledger client, cfg config, firstPerson string) error { + request := &ledgerv1.AppendJournalRequest{ + SourceService: "gl-redistribution-loadtest", + IdempotencyKey: "redistribution:" + cfg.RunID + ":fund:v1", + SourceTransactionId: cfg.RunID + "-fund", + TrackingCode: cfg.RunID + "-fund", + EffectKind: "load-test-funding", + EventVersion: 1, + OccurredAt: timestamppb.Now(), + ActorId: cfg.RunID, + Metadata: map[string]string{"load_test": cfg.RunID, "phase": "funding"}, + Entries: []*ledgerv1.JournalEntry{ + {LineNumber: 1, Account: treasuryAccount(cfg.RunID, cfg.AssetID), Amount: "-10000", Description: "load-test source"}, + {LineNumber: 2, Account: personAccount(firstPerson, cfg.AssetID), Amount: "10000", Description: "initial person balance"}, + }, + } + callCtx, cancel := context.WithTimeout(ctx, cfg.RequestTimeout) + defer cancel() + _, err := ledger.AppendJournal(callCtx, request) + return err +} + +func runRandomTransfers(ctx context.Context, ledger client, cfg config, people [personCount]string, state *balanceState, stats *metrics, output io.Writer) error { + if cfg.Transactions == seedTransactions { + return nil + } + workerCtx, cancel := context.WithCancel(ctx) + defer cancel() + + var sequence atomic.Uint64 + sequence.Store(seedTransactions) + var workers sync.WaitGroup + var firstErr error + var errorOnce sync.Once + workers.Add(cfg.Concurrency) + for worker := range cfg.Concurrency { + go func(worker int) { + defer workers.Done() + random := mathrand.New(mathrand.NewSource(cfg.RandomSeed + int64(worker+1)*7919)) + for workerCtx.Err() == nil { + number := sequence.Add(1) + if number > cfg.Transactions { + return + } + value := state.reserve(random) + started := time.Now() + err := appendTransfer(workerCtx, ledger, cfg, number, people, value) + stats.observe(time.Since(started), err) + if err != nil { + state.rollback(value) + errorOnce.Do(func() { + firstErr = fmt.Errorf("transfer %d: %w", number, err) + cancel() + }) + return + } + succeeded := stats.succeeded.Load() + if succeeded%1_000_000 == 0 { + fmt.Fprintf(output, "progress: %d/%d transfers\n", succeeded, cfg.Transactions) + } + } + }(worker) + } + workers.Wait() + if firstErr != nil { + return firstErr + } + if err := ctx.Err(); err != nil { + return err + } + if succeeded := stats.succeeded.Load(); succeeded != cfg.Transactions { + return fmt.Errorf("completed %d transfers, want %d", succeeded, cfg.Transactions) + } + return nil +} + +func (s *balanceState) reserve(random *mathrand.Rand) transfer { + s.mu.Lock() + defer s.mu.Unlock() + + eligible := make([]int, 0, personCount) + for index, balance := range s.balances { + if balance > minimumBalance { + eligible = append(eligible, index) + } + } + sender := eligible[random.Intn(len(eligible))] + receiver := random.Intn(personCount - 1) + if receiver >= sender { + receiver++ + } + spendable := s.balances[sender] - minimumBalance + amount := random.Int63n(spendable) + 1 + s.balances[sender] -= amount + s.balances[receiver] += amount + return transfer{Sender: sender, Receiver: receiver, Amount: amount} +} + +func (s *balanceState) rollback(value transfer) { + s.mu.Lock() + defer s.mu.Unlock() + s.balances[value.Sender] += value.Amount + s.balances[value.Receiver] -= value.Amount +} + +func appendTransfer(ctx context.Context, ledger client, cfg config, sequence uint64, people [personCount]string, value transfer) error { + number := strconv.FormatUint(sequence, 10) + amount := strconv.FormatInt(value.Amount, 10) + transactionID := cfg.RunID + "-transfer-" + number + request := &ledgerv1.AppendJournalRequest{ + SourceService: "gl-redistribution-loadtest", + IdempotencyKey: "redistribution:" + cfg.RunID + ":transfer:" + number + ":v1", + SourceTransactionId: transactionID, + TrackingCode: transactionID, + EffectKind: "random-internal-transfer", + EventVersion: 1, + OccurredAt: timestamppb.Now(), + CorrelationId: cfg.RunID, + ActorId: people[value.Sender], + Metadata: map[string]string{"load_test": cfg.RunID, "sequence": number}, + Entries: []*ledgerv1.JournalEntry{ + {LineNumber: 1, Account: personAccount(people[value.Sender], cfg.AssetID), Amount: "-" + amount, Description: "random transfer debit"}, + {LineNumber: 2, Account: personAccount(people[value.Receiver], cfg.AssetID), Amount: amount, Description: "random transfer credit"}, + }, + } + callCtx, cancel := context.WithTimeout(ctx, cfg.RequestTimeout) + defer cancel() + _, err := ledger.AppendJournal(callCtx, request) + return err +} + +func personAccount(ownerID string, assetID int64) *ledgerv1.AccountReference { + return &ledgerv1.AccountReference{ + AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE, + OwnerType: "user", + OwnerId: ownerID, + AssetId: assetID, + } +} + +func treasuryAccount(runID string, assetID int64) *ledgerv1.AccountReference { + return &ledgerv1.AccountReference{ + AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_TREASURY, + OwnerType: "load-test", + OwnerId: runID, + AssetId: assetID, + } +} + +func (m *metrics) observe(latency time.Duration, err error) { + m.attempted.Add(1) + m.latencyNanos.Add(latency.Nanoseconds()) + for { + current := m.maxNanos.Load() + if latency.Nanoseconds() <= current || m.maxNanos.CompareAndSwap(current, latency.Nanoseconds()) { + break + } + } + if err != nil { + m.errors.Add(1) + return + } + m.succeeded.Add(1) +} + +func printSummary(output io.Writer, stats *metrics, elapsed time.Duration) { + attempted := stats.attempted.Load() + average := time.Duration(0) + if attempted > 0 { + average = time.Duration(stats.latencyNanos.Load() / int64(attempted)) + } + rate := float64(0) + if elapsed > 0 { + rate = float64(stats.succeeded.Load()) / elapsed.Seconds() + } + fmt.Fprintf(output, "transfers: attempted=%d succeeded=%d errors=%d elapsed=%s rate=%.1f tx/s avg=%s max=%s\n", + attempted, stats.succeeded.Load(), stats.errors.Load(), elapsed.Round(time.Millisecond), rate, average, time.Duration(stats.maxNanos.Load())) +} diff --git a/cmd/gl-redistribution-loadtest/main_test.go b/cmd/gl-redistribution-loadtest/main_test.go new file mode 100644 index 0000000..5500693 --- /dev/null +++ b/cmd/gl-redistribution-loadtest/main_test.go @@ -0,0 +1,150 @@ +package main + +import ( + "bytes" + "context" + "fmt" + mathrand "math/rand" + "strconv" + "strings" + "sync" + "testing" + "time" + + basev1 "gl/gen/base/v1" + ledgerv1 "gl/gen/ledger/v1" + + "google.golang.org/grpc" +) + +type memoryClient struct { + mu sync.Mutex + balances map[string]int64 + journals int +} + +func newMemoryClient() *memoryClient { + return &memoryClient{balances: make(map[string]int64)} +} + +func (c *memoryClient) Health(context.Context, *basev1.Empty, ...grpc.CallOption) (*ledgerv1.HealthResponse, error) { + return &ledgerv1.HealthResponse{Serving: true, DatabaseReady: true}, nil +} + +func (c *memoryClient) AppendJournal(_ context.Context, request *ledgerv1.AppendJournalRequest, _ ...grpc.CallOption) (*ledgerv1.AppendJournalResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + for _, entry := range request.GetEntries() { + amount, err := strconv.ParseInt(entry.GetAmount(), 10, 64) + if err != nil { + return nil, err + } + c.balances[accountKey(entry.GetAccount())] += amount + } + c.journals++ + return &ledgerv1.AppendJournalResponse{Journal: &ledgerv1.Journal{JournalId: fmt.Sprintf("journal-%d", c.journals)}}, nil +} + +func (c *memoryClient) GetBalance(_ context.Context, request *ledgerv1.GetBalanceRequest, _ ...grpc.CallOption) (*ledgerv1.GetBalanceResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + return &ledgerv1.GetBalanceResponse{Account: request.GetAccount(), Balance: strconv.FormatInt(c.balances[accountKey(request.GetAccount())], 10)}, nil +} + +func accountKey(account *ledgerv1.AccountReference) string { + return fmt.Sprintf("%d/%s/%s/%d", account.GetAccountClass(), account.GetOwnerType(), account.GetOwnerId(), account.GetAssetId()) +} + +func TestResolveAssetIDPromptsWhenMissing(t *testing.T) { + var output bytes.Buffer + assetID, err := resolveAssetID(0, strings.NewReader("42\n"), &output) + if err != nil { + t.Fatal(err) + } + if assetID != 42 || output.String() != "Asset number: " { + t.Fatalf("asset=%d prompt=%q", assetID, output.String()) + } +} + +func TestResolveAssetIDUsesFlagWithoutPrompt(t *testing.T) { + var output bytes.Buffer + assetID, err := resolveAssetID(7, strings.NewReader(""), &output) + if err != nil { + t.Fatal(err) + } + if assetID != 7 || output.Len() != 0 { + t.Fatalf("asset=%d prompt=%q", assetID, output.String()) + } +} + +func TestRandomReservationsPreserveSupplyAndMinimum(t *testing.T) { + state := &balanceState{balances: [personCount]int64{9595, 45, 45, 45, 45, 45, 45, 45, 45, 45}} + random := mathrand.New(mathrand.NewSource(1234)) + distinctAmounts := make(map[int64]struct{}) + for range 100_000 { + value := state.reserve(random) + distinctAmounts[value.Amount] = struct{}{} + if value.Sender == value.Receiver || value.Amount <= 0 { + t.Fatalf("invalid transfer: %+v", value) + } + } + + var total int64 + for index, balance := range state.balances { + if balance < minimumBalance { + t.Fatalf("person %d balance %d is below minimum", index+1, balance) + } + total += balance + } + if total != initialSupply { + t.Fatalf("total=%d, want %d", total, initialSupply) + } + if len(distinctAmounts) < 2 { + t.Fatalf("expected random values, got %v", distinctAmounts) + } +} + +func TestVerifyInitialAndFinalBalances(t *testing.T) { + initial := [personCount]int64{10_000} + if err := verifyInitial(initial); err != nil { + t.Fatal(err) + } + final := [personCount]int64{9595, 45, 45, 45, 45, 45, 45, 45, 45, 45} + if err := verifyFinal(final); err != nil { + t.Fatal(err) + } + final[9] = 44 + if err := verifyFinal(final); err == nil { + t.Fatal("expected minimum-balance failure") + } +} + +func TestPersonIDsAreIsolatedByRun(t *testing.T) { + first := personIDs("run-a") + second := personIDs("run-b") + if first[0] == second[0] || first[0] == first[1] { + t.Fatalf("IDs are not isolated: %q %q %q", first[0], second[0], first[1]) + } +} + +func TestExecuteFundsRedistributesAndVerifiesConservation(t *testing.T) { + ledger := newMemoryClient() + var output bytes.Buffer + err := execute(context.Background(), ledger, config{ + RunID: "test-run", + AssetID: 17, + Transactions: 1000, + Concurrency: 4, + RequestTimeout: time.Second, + RandomSeed: 99, + }, &output) + if err != nil { + t.Fatal(err) + } + if ledger.journals != 1001 { + t.Fatalf("journals=%d, want one funding journal and 1000 transfers", ledger.journals) + } + if !strings.Contains(output.String(), "PASS: all 10 people have at least 45; total=10000") { + t.Fatalf("missing successful verification in output:\n%s", output.String()) + } +}