Skip to content

Durable Emitter: Load Testing Updates #23001

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

Open
tarcisiozf wants to merge 28 commits into
base: develop
Choose a base branch
from
Open

Durable Emitter: Load Testing Updates #23001

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
3180a8b
set timeout
tarcisiozf Jul 1, 2026
5e2e125
Fix batch emitter cfg
DylanTinianov Jul 3, 2026
3d3e8ce
Build
DylanTinianov Jul 3, 2026
4fe4f10
Update application.go
DylanTinianov Jul 3, 2026
43882a3
Update application.go
DylanTinianov Jul 3, 2026
0242257
Update application.go
DylanTinianov Jul 6, 2026
b04ee46
Bump common
DylanTinianov Jul 6, 2026
de7b6df
Merge branch 'develop' of github.com:smartcontractkit/chainlink into …
DylanTinianov Jul 6, 2026
d3e0ecd
Bump common
DylanTinianov Jul 6, 2026
1c9787d
synthetic load generator
patrickhuie19 Jul 1, 2026
adb6878
Bump common
DylanTinianov Jul 6, 2026
df148e3
Bump
DylanTinianov Jul 6, 2026
f78082e
Remove synthetic load emitter
DylanTinianov Jul 6, 2026
3c313a5
Bump common
DylanTinianov Jul 7, 2026
3f667ff
Bump common sql skip locked
DylanTinianov Jul 7, 2026
e09dee4
Update application.go
DylanTinianov Jul 7, 2026
4982533
Merge branches 'durable-emitter/set-timeout' and 'develop' of github.…
DylanTinianov Jul 7, 2026
e8b840e
Bump common
DylanTinianov Jul 7, 2026
a6ba317
Generate
DylanTinianov Jul 7, 2026
912b2ff
Remove fallback for drain
DylanTinianov Jul 7, 2026
72ede7e
500 events/s resend
DylanTinianov Jul 7, 2026
4a7711a
Bump common (Retransmit Page Cursor)
DylanTinianov Jul 8, 2026
839e7a3
Reset Autovacuum
DylanTinianov Jul 8, 2026
52deb3d
Update application.go
DylanTinianov Jul 8, 2026
a5ade6f
Bump common - remove noisy logs
DylanTinianov Jul 8, 2026
acd3409
Remove autovacuum migrations
DylanTinianov Jul 8, 2026
e87654a
Bump common
DylanTinianov Jul 8, 2026
d39e2ee
Generate
DylanTinianov Jul 8, 2026
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
Binary file modified core/capabilities/compute/test/fetch/cmd/testdata/output.wasm.br
View file
Open in desktop
Binary file not shown.
Binary file modified core/capabilities/compute/test/simple/cmd/testdata/output.wasm.br
View file
Open in desktop
Binary file not shown.
2 changes: 1 addition & 1 deletion core/scripts/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ require (
github.com/smartcontractkit/chain-selectors v1.0.104
github.com/smartcontractkit/chainlink-automation v0.8.1
github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-common/keystore v1.2.0
github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a
github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
Expand Down
4 changes: 2 additions & 2 deletions core/scripts/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

34 changes: 31 additions & 3 deletions core/services/chainlink/application.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -381,6 +381,29 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
if cfg.Telemetry().DurableEmitterEnabled() && cfg.Telemetry().ChipIngressEndpoint() != "" {
emitterCfg := durableemitter.DefaultConfig()
emitterCfg.Metrics = &durableemitter.DurableEmitterMetricsConfig{}
emitterCfg.InsertBatchSize = 500
emitterCfg.InsertBatchWorkers = 6
emitterCfg.InsertBatchFlushInterval = 100 * time.Millisecond
emitterCfg.DeleteBatchSize = 500
emitterCfg.DeleteBatchWorkers = 6
emitterCfg.DeleteBatchFlushInterval = 100 * time.Millisecond
// Recovery profile: retransmit replays the post-outage backlog while the
// primary path keeps delivering live emits — both feed the same BatchEmitter,
// which has ample headroom (BatchSize × MaxConcurrentSends) for live + backlog.
// - RetransmitInterval = 1s, so RetransmitBatchSize is literally the backlog
// replay rate in events/s. Keep it below BatchEmitter capacity minus the
// live rate, so live emits are never starved out of the buffer.
// - RetransmitAfter MUST exceed delivery latency so an in-flight event is
// never re-queued (the real storm guard — the rate itself is safe because
// MaxConcurrentSends bounds actual downstream load).
// - EventTTL MUST exceed the recovery duration, or the backlog EXPIRES
// (rows dropped by the 1-min expiry loop) before it can be delivered — a
// fake "drain". 6h gives a wide margin for the test.
emitterCfg.RetransmitAfter = 60 * time.Second // > PublishTimeout/MaxPublishTimeout (10s) + buffering
emitterCfg.RetransmitBatchSize = 500 // 500 events/s replayed (interval = 1s)
emitterCfg.RetransmitInterval = 1 * time.Second
emitterCfg.EventTTL = 1 * time.Hour
emitterCfg.PublishTimeout = 10 * time.Second
durableCfg := durableemitter.SetupConfig{
Endpoint: cfg.Telemetry().ChipIngressEndpoint(),
InsecureConnection: cfg.Telemetry().ChipIngressInsecureConnection(),
Expand All @@ -390,9 +413,14 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
AuthPublicKeyHex: csaPubKeyHex,
AuthKeySigner: csaKeystore,
},
RetransmitEnabled: true, // host process owns retransmit
EmitterConfig: &emitterCfg,
Meter: meter,
RetransmitEnabled: true, // host process owns retransmit
EmitterConfig: &emitterCfg,
Meter: meter,
MaxPublishTimeout: 10 * time.Second, // batch emitter rpc timeout
BatchSize: 1000,
BatchInterval: 100 * time.Millisecond,
MaxConcurrentSends: 8,
MessageBufferSize: 50_000,
}
pgStore := durableemitter.NewPgDurableEventStore(opts.DS)
durableEmitter, setupErr := durableemitter.Setup(pgStore, durableCfg, globalLogger)
Expand Down
20 changes: 9 additions & 11 deletions core/services/durableemitter/durable_event_store_orm_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
require.NoError(t, err)
require.Positive(t, id)

events, err := store.ListPending(ctx, time.Now().Add(time.Second), 10)

Check failure on line 37 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 37 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending

Check failure on line 37 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
require.Len(t, events, 1)
assert.Equal(t, id, events[0].ID)
Expand All @@ -42,7 +42,7 @@

require.NoError(t, store.Delete(ctx, id))

events, err = store.ListPending(ctx, time.Now().Add(time.Second), 10)

Check failure on line 45 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 45 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending

Check failure on line 45 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
assert.Empty(t, events)
}
Expand All @@ -57,12 +57,12 @@
require.NoError(t, err)

// createdBefore in the past should return nothing (event was just created).
events, err := store.ListPending(ctx, time.Now().Add(-time.Hour), 10)

Check failure on line 60 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 60 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending

Check failure on line 60 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
assert.Empty(t, events)

// createdBefore in the future should return the event.
events, err = store.ListPending(ctx, time.Now().Add(time.Hour), 10)

Check failure on line 65 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 65 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
assert.Len(t, events, 1)
}
Expand All @@ -78,7 +78,7 @@
require.NoError(t, err)
}

events, err := store.ListPending(ctx, time.Now().Add(time.Second), 5)

Check failure on line 81 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 81 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
assert.Len(t, events, 5)
}
Expand Down Expand Up @@ -122,7 +122,7 @@
assert.Positive(t, st.OldestPendingAge)
}

func TestPgDurableEventStore_MarkDeliveredAndPurgeDelivered(t *testing.T) {
func TestPgDurableEventStore_BatchDelete(t *testing.T) {
db := pgtest.NewSqlxDB(t)
truncateChipDurableEvents(t, db)
ctx := t.Context()
Expand All @@ -131,25 +131,23 @@
id, err := store.Insert(ctx, []byte("payload"))
require.NoError(t, err)

pending, err := store.ListPending(ctx, time.Now().Add(time.Hour), 10)

Check failure on line 134 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 134 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
require.Len(t, pending, 1)

require.NoError(t, store.MarkDelivered(ctx, id))
require.NoError(t, store.MarkDelivered(ctx, id), "second mark is idempotent")
n, err := store.BatchDelete(ctx, []int64{id})
require.NoError(t, err)
require.Equal(t, int64(1), n)

n, err = store.BatchDelete(ctx, []int64{id})
require.NoError(t, err)
require.Equal(t, int64(0), n, "second delete is idempotent")

pending, err = store.ListPending(ctx, time.Now().Add(time.Hour), 10)

Check failure on line 146 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_race_tests) 

not enough arguments in call to store.ListPending

Check failure on line 146 in core/services/durableemitter/durable_event_store_orm_test.go

View workflow job for this annotation

GitHub Actions / Core Tests (go_core_tests) 

not enough arguments in call to store.ListPending
require.NoError(t, err)
require.Empty(t, pending)

var cnt int64
require.NoError(t, db.GetContext(ctx, &cnt, `SELECT count(*) FROM cre.chip_durable_events`))
require.Equal(t, int64(1), cnt, "row remains as tombstone until purge")

n, err := store.PurgeDelivered(ctx, 10)
require.NoError(t, err)
require.Equal(t, int64(1), n)

require.NoError(t, db.GetContext(ctx, &cnt, `SELECT count(*) FROM cre.chip_durable_events`))
require.Equal(t, int64(0), cnt)
require.Equal(t, int64(0), cnt, "row is deleted on delivery, not tombstoned")
}
Binary file modified core/services/workflows/cmd/cre/examples/legacy/data_feeds/testdata/output.wasm.br
View file
Open in desktop
Binary file not shown.
Binary file modified core/services/workflows/test/break/cmd/testdata/output.wasm.br
View file
Open in desktop
Binary file not shown.
Binary file modified core/services/workflows/test/wasm/legacy/cmd/testdata/output.wasm.br
View file
Open in desktop
Binary file not shown.
Binary file modified core/services/workflows/test/zerotimeout/cmd/testdata/output.wasm.br
View file
Open in desktop
Binary file not shown.
2 changes: 1 addition & 1 deletion deployment/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ require (
github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-common/keystore v1.2.0
github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a
github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
Expand Down
4 changes: 2 additions & 2 deletions deployment/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 8 additions & 0 deletions go.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ flowchart LR
chainlink-common --> chainlink-protos/billing/go
chainlink-common --> chainlink-protos/cre/go
chainlink-common --> chainlink-protos/linking-service/go
chainlink-common --> chainlink-protos/metering/go
chainlink-common --> chainlink-protos/node-platform
chainlink-common --> chainlink-protos/storage-service
chainlink-common --> freeport
Expand Down Expand Up @@ -134,6 +135,8 @@ flowchart LR
click chainlink-protos/job-distributor href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/linking-service/go
click chainlink-protos/linking-service/go href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/metering/go
click chainlink-protos/metering/go href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/node-platform
click chainlink-protos/node-platform href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/op-catalog
Expand Down Expand Up @@ -252,6 +255,7 @@ flowchart LR
chainlink-protos/data-feeds
chainlink-protos/job-distributor
chainlink-protos/linking-service/go
chainlink-protos/metering/go
chainlink-protos/node-platform
chainlink-protos/op-catalog
chainlink-protos/orchestrator
Expand Down Expand Up @@ -359,6 +363,7 @@ flowchart LR
chainlink-common --> chainlink-protos/billing/go
chainlink-common --> chainlink-protos/cre/go
chainlink-common --> chainlink-protos/linking-service/go
chainlink-common --> chainlink-protos/metering/go
chainlink-common --> chainlink-protos/node-platform
chainlink-common --> chainlink-protos/storage-service
chainlink-common --> freeport
Expand Down Expand Up @@ -425,6 +430,8 @@ flowchart LR
click chainlink-protos/job-distributor href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/linking-service/go
click chainlink-protos/linking-service/go href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/metering/go
click chainlink-protos/metering/go href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/node-platform
click chainlink-protos/node-platform href "https://github.com/smartcontractkit/chainlink-protos"
chainlink-protos/op-catalog
Expand Down Expand Up @@ -747,6 +754,7 @@ flowchart LR
chainlink-protos/data-feeds
chainlink-protos/job-distributor
chainlink-protos/linking-service/go
chainlink-protos/metering/go
chainlink-protos/node-platform
chainlink-protos/op-catalog
chainlink-protos/orchestrator
Expand Down
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ require (
github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260415165642-49f23e4d76cc
github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260415165642-49f23e4d76cc
github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-common/keystore v1.2.0
github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62
github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion integration-tests/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ require (
github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-common/keystore v1.2.0
github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae
Expand Down
4 changes: 2 additions & 2 deletions integration-tests/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion integration-tests/load/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ require (
github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae
github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.6-0.20260630120514-36abe27604df
Expand Down
4 changes: 2 additions & 2 deletions integration-tests/load/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion system-tests/lib/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ require (
github.com/smartcontractkit/chain-selectors v1.0.104
github.com/smartcontractkit/chainlink-aptos v0.0.0-20260706100550-d43558069754
github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-common/keystore v1.2.0
github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae
Expand Down
4 changes: 2 additions & 2 deletions system-tests/lib/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion system-tests/tests/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ require (
github.com/rs/zerolog v1.35.1
github.com/smartcontractkit/chain-selectors v1.0.104
github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260708174154-2d88e660316e
github.com/smartcontractkit/chainlink-common/keystore v1.2.0
github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501
Expand Down
Loading
Loading