From 6e06d2a9353b8a960c88870a1f00317cf756f06d Mon Sep 17 00:00:00 2001 From: Rodrigo Date: Mon, 6 Jul 2026 11:10:46 -0300 Subject: [PATCH 1/3] feat(load): Add load metrics --- ccip/devenv/impl.go | 6 +- ccip/devenv/tests/load/gun.go | 170 ++++++++++++--- ccip/devenv/tests/load/gun_canton2evm_test.go | 7 +- .../tests/load/gun_canton2evm_token_test.go | 7 +- ccip/devenv/tests/load/gun_evm2canton_test.go | 7 +- .../tests/load/gun_evm2canton_token_test.go | 7 +- ccip/devenv/tests/load/load_helpers.go | 116 ++++++++++- ccip/devenv/tests/load/metrics.go | 193 ++++++++++++++++++ 8 files changed, 467 insertions(+), 46 deletions(-) create mode 100644 ccip/devenv/tests/load/metrics.go diff --git a/ccip/devenv/impl.go b/ccip/devenv/impl.go index 612b1abcf..560ccea07 100644 --- a/ccip/devenv/impl.go +++ b/ccip/devenv/impl.go @@ -1400,7 +1400,6 @@ func (c *Chain) SendMessage(ctx context.Context, dest uint64, fields cciptestint if err != nil { return cciptestinterfaces.MessageSentEvent{}, fmt.Errorf("execute CCIP Send: %w", err) } - c.logger.Info().Str("UpdateID", ccipSendReport.Output.ExecInfo.UpdateID).Msg("CCIP Send executed") update, err := participant.LedgerServices.Update.GetUpdateById(ctx, &apiv2.GetUpdateByIdRequest{ UpdateId: ccipSendReport.Output.ExecInfo.UpdateID, UpdateFormat: &apiv2.UpdateFormat{ @@ -1443,6 +1442,11 @@ func (c *Chain) SendMessage(ctx context.Context, dest uint64, fields cciptestint if err != nil { return cciptestinterfaces.MessageSentEvent{}, err } + c.logger.Info(). + Str("UpdateID", ccipSendReport.Output.ExecInfo.UpdateID). + Str("messageID", protocol.Bytes32(parsedSend.messageID).String()). + Uint64("seqNo", parsedSend.seqNo). + Msg("CCIP Send executed") // Set next holdings err = c.setNextHoldings(update.GetTransaction().GetEvents(), hasTokenTransfer, fields.TokenAmount.Amount) diff --git a/ccip/devenv/tests/load/gun.go b/ccip/devenv/tests/load/gun.go index afddfb8df..a975d4a49 100644 --- a/ccip/devenv/tests/load/gun.go +++ b/ccip/devenv/tests/load/gun.go @@ -1,13 +1,14 @@ package load import ( + "context" "fmt" "sync" "sync/atomic" + "testing" "time" - "github.com/stretchr/testify/require" - + ccv "github.com/smartcontractkit/chainlink-ccv/build/devenv" "github.com/smartcontractkit/chainlink-ccv/build/devenv/cciptestinterfaces" "github.com/smartcontractkit/chainlink-ccv/protocol" "github.com/smartcontractkit/chainlink-testing-framework/wasp" @@ -15,6 +16,23 @@ import ( devenvtests "github.com/smartcontractkit/chainlink-canton/ccip/devenv/tests" ) +// ConfirmSendFunc confirms a CCIP send on the source chain after SendMessage returns. +// TODO: this is needed because EVM impls method has a bug on prod-testnet checks. +type ConfirmSendFunc func( + t *testing.T, + ctx context.Context, + destSelector uint64, + seqNo uint64, + sendResult cciptestinterfaces.MessageSentEvent, +) (cciptestinterfaces.MessageSentEvent, error) + +// LoadGunOptions configures send confirmation and exec timeout for CCIPLoadGun. +type LoadGunOptions struct { + ConfirmSend ConfirmSendFunc + ConfirmExecTimeout time.Duration + SkipExecConfirm bool +} + type loadMessageBuilder func( source cciptestinterfaces.CCIP17, callNum int64, @@ -52,9 +70,11 @@ func (d Destination) BuildMessage( // required; staging/prod runners rely on pre-existing funded accounts. type CCIPLoadGun struct { mu sync.Mutex + flightReady sync.Cond inFlight int32 maxConcurrent int32 calls atomic.Int64 + messageIDs []protocol.Bytes32 source cciptestinterfaces.CCIP17 destinations []Destination @@ -63,8 +83,11 @@ type CCIPLoadGun struct { ccvAddr protocol.UnknownAddress executorAddr protocol.UnknownAddress - confirmSendTimeout time.Duration + confirmSend ConfirmSendFunc confirmExecTimeout time.Duration + skipExecConfirm bool + + metricsCollector *LoadMetricsCollector } // NewCCIPLoadGun wires a CCIP source with one or more destinations for load testing. @@ -72,11 +95,14 @@ func NewCCIPLoadGun( source cciptestinterfaces.CCIP17, destinations []Destination, ccvAddr, executorAddr protocol.UnknownAddress, - confirmExecTimeout time.Duration, + opts LoadGunOptions, ) (*CCIPLoadGun, error) { if source == nil { return nil, fmt.Errorf("CCIPLoadGun: source is nil") } + if opts.ConfirmSend == nil { + return nil, fmt.Errorf("CCIPLoadGun: ConfirmSend is nil") + } if len(destinations) == 0 { return nil, fmt.Errorf("CCIPLoadGun: at least one destination is required") } @@ -95,18 +121,50 @@ func NewCCIPLoadGun( return nil, fmt.Errorf("CCIPLoadGun: destination[%d] mixes token and message-only destinations", i) } } + confirmExecTimeout := opts.ConfirmExecTimeout if confirmExecTimeout <= 0 { confirmExecTimeout = 5 * time.Minute } - return &CCIPLoadGun{ + g := &CCIPLoadGun{ source: source, destinations: destinations, ccvAddr: ccvAddr, executorAddr: executorAddr, - confirmSendTimeout: 30 * time.Second, + confirmSend: opts.ConfirmSend, confirmExecTimeout: confirmExecTimeout, - }, nil + skipExecConfirm: opts.SkipExecConfirm, + metricsCollector: &LoadMetricsCollector{}, + } + g.flightReady.L = &g.mu + + return g, nil +} + +func (g *CCIPLoadGun) acquireSingleFlight() { + g.mu.Lock() + for g.inFlight >= 1 { + g.flightReady.Wait() + } + g.inFlight++ + if g.inFlight > g.maxConcurrent { + g.maxConcurrent = g.inFlight + } + g.mu.Unlock() +} + +func (g *CCIPLoadGun) releaseSingleFlight() { + g.mu.Lock() + g.inFlight-- + if g.inFlight == 0 { + g.flightReady.Broadcast() + } + g.mu.Unlock() +} + +// ConfirmExecTimeout returns the exec confirmation timeout configured for this gun. +func (g *CCIPLoadGun) ConfirmExecTimeout() time.Duration { + return g.confirmExecTimeout } func (g *CCIPLoadGun) nextDestination() Destination { @@ -124,26 +182,10 @@ func (g *CCIPLoadGun) Call(gen *wasp.Generator) *wasp.Response { } t := gen.Cfg.T - var depth int32 - g.mu.Lock() - g.inFlight++ - depth = g.inFlight - if depth > g.maxConcurrent { - g.maxConcurrent = depth - } - g.mu.Unlock() + g.acquireSingleFlight() + defer g.releaseSingleFlight() g.calls.Add(1) - defer func() { - g.mu.Lock() - g.inFlight-- - g.mu.Unlock() - }() - - if depth > 1 { - require.FailNow(t, "overlapping CCIPLoadGun.Call", - "expected single-flight; concurrent depth=%d", depth) - } dest := g.nextDestination() destSelector := dest.Chain.ChainSelector() @@ -151,9 +193,11 @@ func (g *CCIPLoadGun) Call(gen *wasp.Generator) *wasp.Response { callNum := g.calls.Load() fields, opts, err := dest.BuildMessage(g.source, callNum, g.ccvAddr, g.executorAddr) if err != nil { + g.metricsCollector.incrementSendFailure() return &wasp.Response{Failed: true, Error: fmt.Sprintf("BuildMessage (dest=%d): %v", destSelector, err), Duration: time.Since(start)} } + sentTime := time.Now() sendRes, err := g.source.SendMessage( subtestCtx, destSelector, @@ -161,34 +205,70 @@ func (g *CCIPLoadGun) Call(gen *wasp.Generator) *wasp.Response { opts, 3, ) + sendDuration := time.Since(sentTime) if err != nil { + g.metricsCollector.incrementSendFailure() return &wasp.Response{Failed: true, Error: fmt.Sprintf("SendMessage (dest=%d): %v", destSelector, err), Duration: time.Since(start)} } if sendRes.Message == nil { + g.metricsCollector.incrementSendFailure() return &wasp.Response{Failed: true, Error: "SendMessage returned nil message", Duration: time.Since(start)} } seqNo := uint64(sendRes.Message.SequenceNumber) - sentEvent, err := g.source.ConfirmSendOnSource( - subtestCtx, - destSelector, - cciptestinterfaces.MessageEventKey{SeqNum: seqNo}, - g.confirmSendTimeout, - ) + confirmSendStart := time.Now() + sentEvent, err := g.confirmSend(t, subtestCtx, destSelector, seqNo, sendRes) + confirmSendDuration := time.Since(confirmSendStart) if err != nil { - return &wasp.Response{Failed: true, Error: fmt.Sprintf("ConfirmSendOnSource (dest=%d): %v", destSelector, err), Duration: time.Since(start)} + g.metricsCollector.incrementConfirmSendFailure() + return &wasp.Response{Failed: true, Error: fmt.Sprintf("ConfirmSend (dest=%d): %v", destSelector, err), Duration: time.Since(start)} } + g.mu.Lock() + g.messageIDs = append(g.messageIDs, sentEvent.MessageID) + g.mu.Unlock() + + ccv.Plog.Info(). + Str("messageID", sentEvent.MessageID.String()). + Uint64("seqNo", seqNo). + Uint64("destSelector", destSelector). + Msg("Load message confirmed on source") + + sourceChain := g.source.ChainSelector() + record := LoadMessageRecord{ + SeqNo: seqNo, + SourceChain: sourceChain, + DestChain: destSelector, + MessageID: sentEvent.MessageID, + SentTime: sentTime, + SendDuration: sendDuration, + ConfirmSendDuration: confirmSendDuration, + } + + if g.skipExecConfirm { + record.TotalDuration = time.Since(sentTime) + g.metricsCollector.appendRecord(record) + return &wasp.Response{ + Failed: false, + StatusCode: "200", + Duration: time.Since(start), + } + } + + confirmExecStart := time.Now() ev, err := dest.Chain.ConfirmExecOnDest( subtestCtx, g.source.ChainSelector(), cciptestinterfaces.MessageEventKey{SeqNum: seqNo, MessageID: sentEvent.MessageID}, g.confirmExecTimeout, ) + confirmExecDuration := time.Since(confirmExecStart) if err != nil { + g.metricsCollector.incrementConfirmExecFailure() return &wasp.Response{Failed: true, Error: fmt.Sprintf("ConfirmExecOnDest (dest=%d): %v", destSelector, err), Duration: time.Since(start)} } if ev.State != cciptestinterfaces.ExecutionStateSuccess { + g.metricsCollector.incrementConfirmExecFailure() return &wasp.Response{ Failed: true, Error: fmt.Sprintf("execution state=%s (dest=%d)", ev.State.String(), destSelector), @@ -197,6 +277,11 @@ func (g *CCIPLoadGun) Call(gen *wasp.Generator) *wasp.Response { } } + record.ConfirmExecDuration = confirmExecDuration + record.ExecutedTime = time.Now() + record.TotalDuration = record.ExecutedTime.Sub(sentTime) + g.metricsCollector.appendRecord(record) + return &wasp.Response{ Failed: false, StatusCode: "200", @@ -215,3 +300,24 @@ func (g *CCIPLoadGun) MaxConcurrentObserved() int32 { func (g *CCIPLoadGun) CallCount() int64 { return g.calls.Load() } + +// Metrics returns a copy of successful load message timing records. +func (g *CCIPLoadGun) Metrics() []LoadMessageRecord { + records, _ := g.metricsCollector.snapshot() + return records +} + +// FailureCounts returns phase failure counters from failed WASP calls. +func (g *CCIPLoadGun) FailureCounts() LoadFailureCounts { + _, failures := g.metricsCollector.snapshot() + return failures +} + +// MessageIDs returns a copy of message IDs collected after successful ConfirmSend. +func (g *CCIPLoadGun) MessageIDs() []protocol.Bytes32 { + g.mu.Lock() + defer g.mu.Unlock() + out := make([]protocol.Bytes32, len(g.messageIDs)) + copy(out, g.messageIDs) + return out +} diff --git a/ccip/devenv/tests/load/gun_canton2evm_test.go b/ccip/devenv/tests/load/gun_canton2evm_test.go index 0233a7b49..5fa890a58 100644 --- a/ccip/devenv/tests/load/gun_canton2evm_test.go +++ b/ccip/devenv/tests/load/gun_canton2evm_test.go @@ -90,9 +90,12 @@ func TestCanton2EVM_Load(t *testing.T) { destinations, ccvAddr, executorAddr, - utilstests.WaitTimeout(t), + LoadGunOptions{ + ConfirmSend: cantonSourceConfirmSend(cantonChain), + ConfirmExecTimeout: utilstests.WaitTimeout(t), + }, ) require.NoError(t, err) - runWASP(t, gun, "canton-load-canton2evm", sched, "message_only") + runWASP(t, gun, "canton-load-canton2evm", sched, "message_only", nil) } diff --git a/ccip/devenv/tests/load/gun_canton2evm_token_test.go b/ccip/devenv/tests/load/gun_canton2evm_token_test.go index df3a80bc9..cb53d4a26 100644 --- a/ccip/devenv/tests/load/gun_canton2evm_token_test.go +++ b/ccip/devenv/tests/load/gun_canton2evm_token_test.go @@ -78,11 +78,14 @@ func TestCanton2EVM_TokenLoad(t *testing.T) { destinations, ccvAddr, executorAddr, - utilstests.WaitTimeout(t), + LoadGunOptions{ + ConfirmSend: cantonSourceConfirmSend(cantonChain), + ConfirmExecTimeout: utilstests.WaitTimeout(t), + }, ) require.NoError(t, err) - runWASP(t, gun, "canton-load-canton2evm-token", sched, "token_transfer") + runWASP(t, gun, "canton-load-canton2evm-token", sched, "token_transfer", nil) receiverBalanceAfter, err := firstDest.Chain.GetTokenBalance(ctx, firstDest.Receiver, destToken) require.NoError(t, err) diff --git a/ccip/devenv/tests/load/gun_evm2canton_test.go b/ccip/devenv/tests/load/gun_evm2canton_test.go index 034633ae5..86fa2fb4f 100644 --- a/ccip/devenv/tests/load/gun_evm2canton_test.go +++ b/ccip/devenv/tests/load/gun_evm2canton_test.go @@ -66,9 +66,12 @@ func TestEVM2Canton_Load(t *testing.T) { []Destination{cantonDest}, ccvAddr, executorAddr, - utilstests.WaitTimeout(t), + LoadGunOptions{ + ConfirmSend: evmSourceConfirmSend(evmChain), // TODO: this confirmation will change on prod-testnet PR + ConfirmExecTimeout: utilstests.WaitTimeout(t), + }, ) require.NoError(t, err) - runWASP(t, gun, "canton-load-evm2canton", loadSchedule(t), "message_only") + runWASP(t, gun, "canton-load-evm2canton", loadSchedule(t), "message_only", nil) } diff --git a/ccip/devenv/tests/load/gun_evm2canton_token_test.go b/ccip/devenv/tests/load/gun_evm2canton_token_test.go index 426e29fed..cd2fc0cc1 100644 --- a/ccip/devenv/tests/load/gun_evm2canton_token_test.go +++ b/ccip/devenv/tests/load/gun_evm2canton_token_test.go @@ -91,11 +91,14 @@ func TestEVM2Canton_TokenLoad(t *testing.T) { []Destination{cantonDest}, ccvAddr, executorAddr, - utilstests.WaitTimeout(t), + LoadGunOptions{ + ConfirmSend: evmSourceConfirmSend(evmChain), // TODO: this confirmation will change on prod-testnet PR + ConfirmExecTimeout: utilstests.WaitTimeout(t), + }, ) require.NoError(t, err) - runWASP(t, gun, "canton-load-evm2canton-token", sched, "token_transfer") + runWASP(t, gun, "canton-load-evm2canton-token", sched, "token_transfer", nil) totalHoldingsRat, err := testhelpers.GetHoldingsBalance(ctx, receiverParticipant, nil) require.NoError(t, err) diff --git a/ccip/devenv/tests/load/load_helpers.go b/ccip/devenv/tests/load/load_helpers.go index 4bf0b1399..6343a2710 100644 --- a/ccip/devenv/tests/load/load_helpers.go +++ b/ccip/devenv/tests/load/load_helpers.go @@ -5,6 +5,7 @@ import ( "fmt" "math/big" "os" + "strings" "testing" "time" @@ -16,6 +17,7 @@ import ( "github.com/smartcontractkit/chainlink-ccv/build/devenv/cciptestinterfaces" "github.com/smartcontractkit/chainlink-ccv/build/devenv/common" ccvload "github.com/smartcontractkit/chainlink-ccv/build/devenv/tests/e2e/load" + ccvmetrics "github.com/smartcontractkit/chainlink-ccv/build/devenv/tests/e2e/metrics" "github.com/smartcontractkit/chainlink-ccv/protocol" "github.com/smartcontractkit/chainlink-deployments-framework/datastore" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/blockchain" @@ -29,10 +31,13 @@ import ( ) const ( - envMessageRate = "CANTON_LOAD_MESSAGE_RATE" - envLoadDuration = "CANTON_LOAD_DURATION" - defaultMessageRate = "1/1s" - defaultLoadDuration = 90 * time.Second + envMessageRate = "CANTON_LOAD_MESSAGE_RATE" + envLoadDuration = "CANTON_LOAD_DURATION" + envLoadCallTimeout = "CANTON_LOAD_CALL_TIMEOUT" + defaultMessageRate = "1/1s" + defaultLoadDuration = 90 * time.Second + defaultLoadCallPadding = 2 * time.Minute + confirmSendTimeout = 30 * time.Second ) type scheduleConfig struct { @@ -70,9 +75,71 @@ func loadSchedule(t *testing.T) scheduleConfig { } } -func runWASP(t *testing.T, gun *CCIPLoadGun, genName string, sched scheduleConfig, scenario string) { +func waspCallTimeout(t *testing.T, gun *CCIPLoadGun, sched scheduleConfig) time.Duration { t.Helper() + if v := strings.TrimSpace(os.Getenv(envLoadCallTimeout)); v != "" { + parsed, err := time.ParseDuration(v) + require.NoError(t, err, "%s=%q invalid", envLoadCallTimeout, v) + return parsed + } + + return gun.ConfirmExecTimeout() + sched.rateUnit + defaultLoadCallPadding +} + +func printLoadMetrics(t *testing.T, gun *CCIPLoadGun) { + t.Helper() + + records := gun.Metrics() + failures := gun.FailureCounts() + PrintPhaseMetricsSummary(t, records, failures, false) + + ccvMetrics := ToCCVMessageMetrics(records) + if len(ccvMetrics) > 0 { + totals := ccvmetrics.MessageTotals{ + Sent: len(ccvMetrics), + Received: len(ccvMetrics), + } + summary := ccvmetrics.CalculateMetricsSummary(ccvMetrics, totals) + ccvmetrics.PrintMetricsSummary(t, summary) + } +} + +func logLoadMessageSummary(t *testing.T, gun *CCIPLoadGun, indexerEndpoints []string) { + t.Helper() + + ids := gun.MessageIDs() + lggr := ccv.Plog + lggr.Info().Int("count", len(ids)).Msg("Load message summary") + + var indexerBase string + if len(indexerEndpoints) > 0 { + indexerBase = strings.TrimSuffix(indexerEndpoints[0], "/") + } + + for i, id := range ids { + msgID := id.String() + ev := lggr.Info().Int("index", i+1).Str("messageID", msgID) + if indexerBase != "" { + ev = ev.Str("indexer", fmt.Sprintf("%s/v1/verifierresults/%s", indexerBase, msgID)) + } + ev.Msg("Load message sent") + } +} + +func runWASP(t *testing.T, gun *CCIPLoadGun, genName string, sched scheduleConfig, scenario string, indexerEndpoints []string) { + t.Helper() + defer logLoadMessageSummary(t, gun, indexerEndpoints) + + callTimeout := waspCallTimeout(t, gun, sched) + ccv.Plog.Info(). + Str("messageRate", sched.messageRate). + Dur("rateUnit", sched.rateUnit). + Dur("loadDuration", sched.duration). + Dur("callTimeout", callTimeout). + Dur("confirmExecTimeout", gun.ConfirmExecTimeout()). + Msg("WASP load schedule") + labels := map[string]string{ "go_test_name": genName, "branch": "test", @@ -92,6 +159,7 @@ func runWASP(t *testing.T, gun *CCIPLoadGun, genName string, sched scheduleConfi wasp.Plain(sched.rate, sched.duration), ), RateLimitUnitDuration: sched.rateUnit, + CallTimeout: callTimeout, Gun: gun, Labels: labels, LokiConfig: nil, @@ -104,6 +172,8 @@ func runWASP(t *testing.T, gun *CCIPLoadGun, genName string, sched scheduleConfi require.Positive(t, gun.CallCount(), "gun should have completed at least one message") require.LessOrEqual(t, gun.MaxConcurrentObserved(), int32(1), "Gun.Call must not overlap (single-flight)") + + printLoadMetrics(t, gun) } func discoverEVMDestinations(t *testing.T, in *ccv.Cfg, chainMap map[uint64]cciptestinterfaces.CCIP17) []Destination { @@ -370,3 +440,39 @@ func resolveEVMSourceAddrs(t *testing.T, lib ccv.Lib, evmSelector uint64) (proto return ccvAddr, executorAddr } + +func cantonSourceConfirmSend(source cciptestinterfaces.CCIP17) ConfirmSendFunc { + return func( + t *testing.T, + ctx context.Context, + destSelector uint64, + seqNo uint64, + _ cciptestinterfaces.MessageSentEvent, + ) (cciptestinterfaces.MessageSentEvent, error) { + return source.ConfirmSendOnSource( + ctx, + destSelector, + cciptestinterfaces.MessageEventKey{SeqNum: seqNo}, + confirmSendTimeout, + ) + } +} + +// TODO: this is needed because EVM impls method has a bug on prod-testnet checks. +// currently just a simple wrapper around the method. +func evmSourceConfirmSend(source cciptestinterfaces.CCIP17) ConfirmSendFunc { + return func( + t *testing.T, + ctx context.Context, + destSelector uint64, + seqNo uint64, + _ cciptestinterfaces.MessageSentEvent, + ) (cciptestinterfaces.MessageSentEvent, error) { + return source.ConfirmSendOnSource( + ctx, + destSelector, + cciptestinterfaces.MessageEventKey{SeqNum: seqNo}, + confirmSendTimeout, + ) + } +} diff --git a/ccip/devenv/tests/load/metrics.go b/ccip/devenv/tests/load/metrics.go new file mode 100644 index 000000000..4d405a0e9 --- /dev/null +++ b/ccip/devenv/tests/load/metrics.go @@ -0,0 +1,193 @@ +package load + +import ( + "slices" + "sync" + "testing" + "time" + + ccvmetrics "github.com/smartcontractkit/chainlink-ccv/build/devenv/tests/e2e/metrics" + "github.com/smartcontractkit/chainlink-ccv/protocol" +) + +// LoadMessageRecord captures per-message phase timings for a successful load test call. +type LoadMessageRecord struct { + SeqNo uint64 + SourceChain uint64 + DestChain uint64 + MessageID protocol.Bytes32 + SentTime time.Time + ExecutedTime time.Time + SendDuration time.Duration + ConfirmSendDuration time.Duration + ConfirmExecDuration time.Duration + TotalDuration time.Duration +} + +// LoadFailureCounts tracks failures by load test phase. +type LoadFailureCounts struct { + Send int + ConfirmSend int + ConfirmExec int +} + +// LoadMetricsCollector accumulates successful message records and phase failure counts. +type LoadMetricsCollector struct { + mu sync.Mutex + records []LoadMessageRecord + failures LoadFailureCounts +} + +func (c *LoadMetricsCollector) appendRecord(r LoadMessageRecord) { + c.mu.Lock() + defer c.mu.Unlock() + c.records = append(c.records, r) +} + +func (c *LoadMetricsCollector) incrementSendFailure() { + c.mu.Lock() + defer c.mu.Unlock() + c.failures.Send++ +} + +func (c *LoadMetricsCollector) incrementConfirmSendFailure() { + c.mu.Lock() + defer c.mu.Unlock() + c.failures.ConfirmSend++ +} + +func (c *LoadMetricsCollector) incrementConfirmExecFailure() { + c.mu.Lock() + defer c.mu.Unlock() + c.failures.ConfirmExec++ +} + +func (c *LoadMetricsCollector) snapshot() ([]LoadMessageRecord, LoadFailureCounts) { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]LoadMessageRecord, len(c.records)) + copy(out, c.records) + return out, c.failures +} + +// PercentileStats holds common percentile values for a set of durations. +type PercentileStats struct { + Min time.Duration + Max time.Duration + P90 time.Duration + P95 time.Duration + P99 time.Duration +} + +func calculatePercentiles(durations []time.Duration) PercentileStats { + if len(durations) == 0 { + return PercentileStats{} + } + + slices.Sort(durations) + + p90Index := int(float64(len(durations)) * 0.90) + p95Index := int(float64(len(durations)) * 0.95) + p99Index := int(float64(len(durations)) * 0.99) + + if p90Index >= len(durations) { + p90Index = len(durations) - 1 + } + if p95Index >= len(durations) { + p95Index = len(durations) - 1 + } + if p99Index >= len(durations) { + p99Index = len(durations) - 1 + } + + return PercentileStats{ + Min: durations[0], + Max: durations[len(durations)-1], + P90: durations[p90Index], + P95: durations[p95Index], + P99: durations[p99Index], + } +} + +func logPhasePercentiles(t *testing.T, label string, stats PercentileStats, count int) { + t.Helper() + t.Logf("%s (n=%d):\n"+ + " Min: %v\n"+ + " Max: %v\n"+ + " P90: %v\n"+ + " P95: %v\n"+ + " P99: %v", + label, count, + stats.Min, stats.Max, stats.P90, stats.P95, stats.P99, + ) +} + +// PrintPhaseMetricsSummary prints per-phase timing percentiles for successful load messages. +func PrintPhaseMetricsSummary(t *testing.T, records []LoadMessageRecord, failures LoadFailureCounts, skipExecConfirm bool) { + t.Helper() + + t.Logf("\n" + + "========================================\n" + + " Load Phase Metrics \n" + + "========================================") + + if len(records) == 0 { + t.Logf("No successful load messages recorded") + } else { + sendDurations := make([]time.Duration, len(records)) + confirmSendDurations := make([]time.Duration, len(records)) + for i, r := range records { + sendDurations[i] = r.SendDuration + confirmSendDurations[i] = r.ConfirmSendDuration + } + + logPhasePercentiles(t, "Send", calculatePercentiles(sendDurations), len(records)) + logPhasePercentiles(t, "Confirm Send", calculatePercentiles(confirmSendDurations), len(records)) + + if skipExecConfirm { + sourceConfirmed := make([]time.Duration, len(records)) + for i, r := range records { + sourceConfirmed[i] = r.SendDuration + r.ConfirmSendDuration + } + logPhasePercentiles(t, "Source Confirmed (Send → Confirm Send)", calculatePercentiles(sourceConfirmed), len(records)) + } else { + confirmExecDurations := make([]time.Duration, len(records)) + totalDurations := make([]time.Duration, len(records)) + for i, r := range records { + confirmExecDurations[i] = r.ConfirmExecDuration + totalDurations[i] = r.TotalDuration + } + logPhasePercentiles(t, "Confirm Exec", calculatePercentiles(confirmExecDurations), len(records)) + logPhasePercentiles(t, "Total (E2E)", calculatePercentiles(totalDurations), len(records)) + } + } + + totalFailures := failures.Send + failures.ConfirmSend + failures.ConfirmExec + if totalFailures > 0 { + t.Logf("----------------------------------------\n"+ + "Failures: send=%d confirm_send=%d confirm_exec=%d", + failures.Send, failures.ConfirmSend, failures.ConfirmExec) + } + + t.Logf("========================================") +} + +// ToCCVMessageMetrics maps load records with ExecutedTime set to CCV message metrics. +func ToCCVMessageMetrics(records []LoadMessageRecord) []ccvmetrics.MessageMetrics { + out := make([]ccvmetrics.MessageMetrics, 0, len(records)) + for _, r := range records { + if r.ExecutedTime.IsZero() { + continue + } + out = append(out, ccvmetrics.MessageMetrics{ + SeqNo: r.SeqNo, + MessageID: r.MessageID.String(), + SourceChain: r.SourceChain, + DestChain: r.DestChain, + SentTime: r.SentTime, + ExecutedTime: r.ExecutedTime, + LatencyDuration: r.ExecutedTime.Sub(r.SentTime), + }) + } + return out +} From c9cfe0365781fb4d01ebd964716743868c4f7a8e Mon Sep 17 00:00:00 2001 From: Rodrigo Date: Tue, 7 Jul 2026 12:04:37 -0300 Subject: [PATCH 2/3] fix(lint) --- ccip/devenv/tests/load/gun.go | 1 + ccip/devenv/tests/load/gun_canton2evm_test.go | 2 +- .../tests/load/gun_canton2evm_token_test.go | 2 +- ccip/devenv/tests/load/gun_evm2canton_test.go | 2 +- .../tests/load/gun_evm2canton_token_test.go | 2 +- ccip/devenv/tests/load/load_helpers.go | 18 ++++-------------- ccip/devenv/tests/load/metrics.go | 2 ++ 7 files changed, 11 insertions(+), 18 deletions(-) diff --git a/ccip/devenv/tests/load/gun.go b/ccip/devenv/tests/load/gun.go index a975d4a49..9f9e94dcb 100644 --- a/ccip/devenv/tests/load/gun.go +++ b/ccip/devenv/tests/load/gun.go @@ -319,5 +319,6 @@ func (g *CCIPLoadGun) MessageIDs() []protocol.Bytes32 { defer g.mu.Unlock() out := make([]protocol.Bytes32, len(g.messageIDs)) copy(out, g.messageIDs) + return out } diff --git a/ccip/devenv/tests/load/gun_canton2evm_test.go b/ccip/devenv/tests/load/gun_canton2evm_test.go index 5fa890a58..29060312c 100644 --- a/ccip/devenv/tests/load/gun_canton2evm_test.go +++ b/ccip/devenv/tests/load/gun_canton2evm_test.go @@ -97,5 +97,5 @@ func TestCanton2EVM_Load(t *testing.T) { ) require.NoError(t, err) - runWASP(t, gun, "canton-load-canton2evm", sched, "message_only", nil) + runWASP(t, gun, "canton-load-canton2evm", sched, "message_only") } diff --git a/ccip/devenv/tests/load/gun_canton2evm_token_test.go b/ccip/devenv/tests/load/gun_canton2evm_token_test.go index cb53d4a26..291426a30 100644 --- a/ccip/devenv/tests/load/gun_canton2evm_token_test.go +++ b/ccip/devenv/tests/load/gun_canton2evm_token_test.go @@ -85,7 +85,7 @@ func TestCanton2EVM_TokenLoad(t *testing.T) { ) require.NoError(t, err) - runWASP(t, gun, "canton-load-canton2evm-token", sched, "token_transfer", nil) + runWASP(t, gun, "canton-load-canton2evm-token", sched, "token_transfer") receiverBalanceAfter, err := firstDest.Chain.GetTokenBalance(ctx, firstDest.Receiver, destToken) require.NoError(t, err) diff --git a/ccip/devenv/tests/load/gun_evm2canton_test.go b/ccip/devenv/tests/load/gun_evm2canton_test.go index 86fa2fb4f..b33555040 100644 --- a/ccip/devenv/tests/load/gun_evm2canton_test.go +++ b/ccip/devenv/tests/load/gun_evm2canton_test.go @@ -73,5 +73,5 @@ func TestEVM2Canton_Load(t *testing.T) { ) require.NoError(t, err) - runWASP(t, gun, "canton-load-evm2canton", loadSchedule(t), "message_only", nil) + runWASP(t, gun, "canton-load-evm2canton", loadSchedule(t), "message_only") } diff --git a/ccip/devenv/tests/load/gun_evm2canton_token_test.go b/ccip/devenv/tests/load/gun_evm2canton_token_test.go index cd2fc0cc1..4ad00269d 100644 --- a/ccip/devenv/tests/load/gun_evm2canton_token_test.go +++ b/ccip/devenv/tests/load/gun_evm2canton_token_test.go @@ -98,7 +98,7 @@ func TestEVM2Canton_TokenLoad(t *testing.T) { ) require.NoError(t, err) - runWASP(t, gun, "canton-load-evm2canton-token", sched, "token_transfer", nil) + runWASP(t, gun, "canton-load-evm2canton-token", sched, "token_transfer") totalHoldingsRat, err := testhelpers.GetHoldingsBalance(ctx, receiverParticipant, nil) require.NoError(t, err) diff --git a/ccip/devenv/tests/load/load_helpers.go b/ccip/devenv/tests/load/load_helpers.go index 6343a2710..0ba7e3daa 100644 --- a/ccip/devenv/tests/load/load_helpers.go +++ b/ccip/devenv/tests/load/load_helpers.go @@ -105,31 +105,21 @@ func printLoadMetrics(t *testing.T, gun *CCIPLoadGun) { } } -func logLoadMessageSummary(t *testing.T, gun *CCIPLoadGun, indexerEndpoints []string) { +func logLoadMessageSummary(t *testing.T, gun *CCIPLoadGun) { t.Helper() ids := gun.MessageIDs() lggr := ccv.Plog lggr.Info().Int("count", len(ids)).Msg("Load message summary") - var indexerBase string - if len(indexerEndpoints) > 0 { - indexerBase = strings.TrimSuffix(indexerEndpoints[0], "/") - } - for i, id := range ids { - msgID := id.String() - ev := lggr.Info().Int("index", i+1).Str("messageID", msgID) - if indexerBase != "" { - ev = ev.Str("indexer", fmt.Sprintf("%s/v1/verifierresults/%s", indexerBase, msgID)) - } - ev.Msg("Load message sent") + lggr.Info().Int("index", i+1).Str("messageID", id.String()).Msg("Load message sent") } } -func runWASP(t *testing.T, gun *CCIPLoadGun, genName string, sched scheduleConfig, scenario string, indexerEndpoints []string) { +func runWASP(t *testing.T, gun *CCIPLoadGun, genName string, sched scheduleConfig, scenario string) { t.Helper() - defer logLoadMessageSummary(t, gun, indexerEndpoints) + defer logLoadMessageSummary(t, gun) callTimeout := waspCallTimeout(t, gun, sched) ccv.Plog.Info(). diff --git a/ccip/devenv/tests/load/metrics.go b/ccip/devenv/tests/load/metrics.go index 4d405a0e9..38e0d7b80 100644 --- a/ccip/devenv/tests/load/metrics.go +++ b/ccip/devenv/tests/load/metrics.go @@ -67,6 +67,7 @@ func (c *LoadMetricsCollector) snapshot() ([]LoadMessageRecord, LoadFailureCount defer c.mu.Unlock() out := make([]LoadMessageRecord, len(c.records)) copy(out, c.records) + return out, c.failures } @@ -189,5 +190,6 @@ func ToCCVMessageMetrics(records []LoadMessageRecord) []ccvmetrics.MessageMetric LatencyDuration: r.ExecutedTime.Sub(r.SentTime), }) } + return out } From 0ea19d2c45b9694946c1a8441b9f7a49f9899cac Mon Sep 17 00:00:00 2001 From: Rodrigo Date: Tue, 7 Jul 2026 16:03:59 -0300 Subject: [PATCH 3/3] fix(e2e): Fix tests holding minting --- ccip/devenv/tests/e2e/canton2evm_e2e_test.go | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/ccip/devenv/tests/e2e/canton2evm_e2e_test.go b/ccip/devenv/tests/e2e/canton2evm_e2e_test.go index 28e2e1a3b..ca7550055 100644 --- a/ccip/devenv/tests/e2e/canton2evm_e2e_test.go +++ b/ccip/devenv/tests/e2e/canton2evm_e2e_test.go @@ -52,11 +52,15 @@ func TestCanton2EVM_Basic(t *testing.T) { require.NoError(t, err) }) + // TODO: currently minting 2 holdings of 2k for all the e2e tests. Otherwise one holding might conflict with the other. + // the tests need to be hardened to not rely on this and instead correctly pick the holding that suffices. + require.NoError(t, cantonImpl.MintTokens(ctx, uint64(devenvtests.CantonToEVMFeeAmount)*100)) + require.NoError(t, cantonImpl.MintTokens(ctx, uint64(devenvtests.CantonToEVMFeeAmount)*100)) + t.Run("EOA receiver and default committee verifier", func(t *testing.T) { subtestCtx := ccv.Plog.WithContext(t.Context()) // Setup message send - require.NoError(t, cantonImpl.MintTokens(ctx, uint64(devenvtests.CantonToEVMFeeAmount))) require.NoError(t, cantonImpl.SetupSend(ctx, uint64(devenvtests.CantonToEVMFeeAmount), 0)) ds, err := lib.DataStore() @@ -146,12 +150,6 @@ func TestCanton2EVM_Basic(t *testing.T) { tokenTransferAmount := lane.TransferAmount.Uint64() // Setup message send - require.NoError(t, cantonImpl.MintTokens(ctx, - devenvtests.CantonToEVMTokenSequentialSends*uint64(devenvtests.CantonToEVMFeeAmount), - )) // Holdings for fee - require.NoError(t, cantonImpl.MintTokens(ctx, - devenvtests.CantonToEVMTokenSequentialSends*tokenTransferAmount, - )) // Holdings for token transfer require.NoError(t, cantonImpl.SetupSend(ctx, uint64(devenvtests.CantonToEVMFeeAmount), tokenTransferAmount)) ds, err := lib.DataStore() @@ -237,8 +235,6 @@ func TestCanton2EVM_Basic(t *testing.T) { lane := devenvtests.ResolveTokenLane(t, in, lib, chainMap, cantonChain.ChainSelector(), []uint64{evmChain.ChainSelector()}) tokenTransferAmount := lane.TransferAmount.Uint64() - require.NoError(t, cantonImpl.MintTokens(ctx, uint64(devenvtests.CantonToEVMFeeAmount))) - require.NoError(t, cantonImpl.MintTokens(ctx, tokenTransferAmount)) require.NoError(t, cantonImpl.SetupSend(ctx, uint64(devenvtests.CantonToEVMFeeAmount), tokenTransferAmount)) receiver, err := evmChain.GetEOAReceiverAddress() @@ -281,8 +277,6 @@ func TestCanton2EVM_Basic(t *testing.T) { lane := devenvtests.ResolveTokenLane(t, in, lib, chainMap, cantonChain.ChainSelector(), []uint64{evmChain.ChainSelector()}) tokenTransferAmount := lane.TransferAmount.Uint64() - require.NoError(t, cantonImpl.MintTokens(ctx, uint64(devenvtests.CantonToEVMFeeAmount))) - require.NoError(t, cantonImpl.MintTokens(ctx, tokenTransferAmount)) require.NoError(t, cantonImpl.SetupSend(ctx, uint64(devenvtests.CantonToEVMFeeAmount), tokenTransferAmount)) ds, err := lib.DataStore()