Batch insert records into the upkeep state store #10488

ferglor · 2023-09-05T13:48:38Z

No description provided.

github-actions · 2023-09-05T13:49:12Z

I see that you haven't updated any CHANGELOG files. Would it make sense to do so?

infiloop2 · 2023-09-05T15:40:03Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/orm.go

+	for i, record := range state {
+		fieldOffset := i * fieldsPerRecord
+		valueStrings = append(valueStrings, fmt.Sprintf(
+			"($%d::NUMERIC, $%d, $%d, $%d, $%d, $%d::NUMERIC, $%d)",
+			fieldOffset+1, fieldOffset+2, fieldOffset+3, fieldOffset+4, fieldOffset+5, fieldOffset+6, fieldOffset+7,
+		))
+		valueArgs = append(valueArgs, o.chainID, record.WorkID, record.CompletionState, record.BlockNumber, record.InsertedAt, record.UpkeepID, record.IneligibilityReason)
+	}
+
+	query := fmt.Sprintf(`
+		INSERT INTO evm_upkeep_states (evm_chain_id, work_id, completion_state, block_number, inserted_at, upkeep_id, ineligibility_reason)
+		VALUES %s
+		ON CONFLICT (evm_chain_id, work_id)
+		DO NOTHING;
+	`, strings.Join(valueStrings, ", "))


I think you might be able to do this more elegantly like https://github.com/smartcontractkit/chainlink/blob/develop/core/chains/evm/logpoller/orm.go#L187-L190

infiloop2 · 2023-09-05T15:40:35Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+	})
+
+	if len(u.pendingRecords) == flushSize {
+		return u.flush(ctx)


should we try to do this in background instead of blocking the 1000th upsertStateRecord?

Sure! So would we want to run the flush in the background, and then continue to collecting more records, but block the next flush until the previous flush completes?

not sure fully, will need to consider what happens during errors

It looks like when we call the post processors in the plugin, we only log out the errors, we don't act on them further. So maybe, if we encounter an error when we flush, we could send the error down a channel and have the Start function read from the channel, and log any errors encountered? So that we still log the error, but just in a different place? Though it probably won't be clear that a post processor triggered the error?

yeah potentially. We also need to consider if we retry flushing it or not, o/w the queue size can keep on growing

maybe the simplest for now is to keep it as is.
One thing i'm not sure if we need to do is change to if len(u.pendingRecords) >= flushSize { ?

infiloop2 · 2023-09-05T15:41:06Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/orm_test.go

@@ -21,16 +21,16 @@ func TestInsertSelectDelete(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	orm := NewORM(chainID, db, lggr, pgtest.NewQConfig(true))

-	inserted := persistedStateRecord{
+	inserted := []persistedStateRecord{{


can we try to test for multiple records

infiloop2

lgtm overall, couple of comments. Would also be good to get @EasterTheBunny / @amirylm 's review

github-actions · 2023-09-07T12:25:01Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6109657253

github-actions · 2023-09-07T12:39:28Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6109812434

github-actions · 2023-09-07T13:51:10Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6110639305

infiloop2 · 2023-09-07T14:39:50Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+	copy(cloneRecords, u.pendingRecords)
+	u.pendingRecords = []persistedStateRecord{}


would we need a lock on u.pendingRecords here? since it can be written to in upsertStateRecord

infiloop2 · 2023-09-07T14:41:04Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+	u.pendingRecords = []persistedStateRecord{}
+
+	go func() {
+		if err := u.orm.BatchInsertUpkeepStates(cloneRecords, pg.WithParentCtx(ctx)); err != nil {


should we enforce some max batch size here?

infiloop2 · 2023-09-07T14:54:59Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+	if len(u.pendingRecords) >= flushSize {
+		u.flush(ctx)
+	}


as discussed offline, might be cleaner to make flush independent of upsertRecords and trigger it on a timer. (while flushing records in batches)

github-actions · 2023-09-08T02:04:20Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6116822671

infiloop2 · 2023-09-08T10:07:24Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+
+	go func() {
+		if err := u.orm.BatchInsertUpkeepStates(cloneRecords, pg.WithParentCtx(ctx)); err != nil {
+			u.errCh <- err


nit. I think you can simply log the error here?

infiloop2 · 2023-09-08T10:08:29Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

@@ -129,6 +146,31 @@ func (u *upkeepStateStore) Start(pctx context.Context) error {
 	return nil
 }

+func (u *upkeepStateStore) flush(ctx context.Context) {
+	u.sem <- struct{}{}


nit. don't fully understand the reason for this, is it to limit at most 10 (channel size) flushes at once?

Yeah I didn't want to have an unbounded number of go routines running (since I'm not sure how frequently we'll need to flush to the DB, or how long those flushes can take) so I just ballparked at most 10 at a time. Happy to remove this if it's overkill!

i think this will automatically be limited since flush is done in a loop synchronously? Don't have strong opinions ok to keep as is

Bit the loop triggers the batch insert asynchronously?

infiloop2 · 2023-09-08T10:10:28Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+	cloneRecords := u.pendingRecords[:batch]
+	u.pendingRecords = u.pendingRecords[batch:]


not sure if we should have a loop here in case pendingRecords grow very fast over time?

will also need to think about the time we hold the lock though, probably not a good idea to hold the lock while waiting on DB insert

github-actions · 2023-09-08T13:13:05Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6122188246

infiloop2 · 2023-09-08T13:31:53Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

 				case <-ctx.Done():
-
+					u.flush(ctx)
+					u.doneCh <- struct{}{}


nit. couldn't find anyone reading from this channel?

Note that I'm making some changes that affects it in https://github.com/smartcontractkit/chainlink/pull/10535/files

@infiloop2 currently the tests read from this to check that the store has stopped before proceeding

infiloop2 · 2023-09-08T13:32:41Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+		cleanCadence:   GCInterval,
+		pendingRecords: []persistedStateRecord{},
+		errCh:          make(chan error, 1),
+		sem:            make(chan struct{}, 10),


nit. might be better to define 10 as a constant

Ideally all values like this should be constants. We can add a linter that catches "magic numbers". It's pretty effective but can get annoying too. ¯_(ツ)_/¯

infiloop2

lgtm!

github-actions · 2023-09-08T15:54:43Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6123827855

amirylm · 2023-09-08T15:55:28Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

@@ -119,8 +134,14 @@ func (u *upkeepStateStore) Start(pctx context.Context) error {
 					}

 					ticker.Reset(utils.WithJitter(u.cleanCadence))
+				case <-flushTicker.C:
+					u.flush(ctx)
+				case err := <-u.errCh:


we write into errCh only from flush, let's also listen to it from there (or maybe avoid it, as we anyway just log the errors)

I've just removed it!

github-actions · 2023-09-08T16:03:19Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6123906384

EasterTheBunny

Looks good. My only request would be to add jitter to one ticker and remove it from the other.

EasterTheBunny · 2023-09-08T17:56:01Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

 			ticker := time.NewTicker(utils.WithJitter(u.cleanCadence))
 			defer ticker.Stop()

+			flushTicker := newTickerFn(flushCadence)
+			defer flushTicker.Stop()


One request that came to me when lasting working on this was to add jitter to the tickers that were making database calls. I don't think first one needs the jitter anymore, but the second one does.

EasterTheBunny · 2023-09-08T17:57:07Z

core/services/ocr2/plugins/ocr2keeper/evm21/upkeepstate/store.go

+		cleanCadence:   GCInterval,
+		pendingRecords: []persistedStateRecord{},
+		errCh:          make(chan error, 1),
+		sem:            make(chan struct{}, 10),


Ideally all values like this should be constants. We can add a linter that catches "magic numbers". It's pretty effective but can get annoying too. ¯_(ツ)_/¯

github-actions · 2023-09-08T18:18:08Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6125104720

github-actions · 2023-09-08T18:39:50Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6125289893

github-actions · 2023-09-08T19:08:17Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6125520311

github-actions · 2023-09-08T19:40:45Z

Running downstream job at https://github.com/smartcontractkit/operator-ui/actions/runs/6125778171

cl-sonarqube-production · 2023-09-08T20:02:11Z

SonarQube Quality Gate

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

96.2% Coverage
0.0% Duplication

ferglor requested a review from a team as a code owner September 5, 2023 13:48

infiloop2 reviewed Sep 5, 2023

View reviewed changes

infiloop2 requested review from EasterTheBunny and amirylm September 5, 2023 15:41

infiloop2 reviewed Sep 5, 2023

View reviewed changes

infiloop2 reviewed Sep 7, 2023

View reviewed changes

ferglor added 4 commits September 7, 2023 23:32

Use a batch insert

231ad07

No longer insert records, store them for bulk insert, fail tests

6e5943b

Synchronous tick inserts

aa6811f

Insert on a tick

36fa612

ferglor force-pushed the feature/AUTO-4895 branch from ea9f6d0 to 36fa612 Compare September 8, 2023 02:03

infiloop2 reviewed Sep 8, 2023

View reviewed changes

ferglor added 3 commits September 8, 2023 12:38

Data race

6982c88

Test flake with inserts not completing

a70e831

Wait for writes

24beece

infiloop2 reviewed Sep 8, 2023

View reviewed changes

infiloop2 previously approved these changes Sep 8, 2023

View reviewed changes

amirylm reviewed Sep 8, 2023

View reviewed changes

Remove the errCh

66823fd

EasterTheBunny requested changes Sep 8, 2023

View reviewed changes

Add jitter to the ticker, remove it from the other

021c387

ferglor requested a review from EasterTheBunny September 8, 2023 18:16

EasterTheBunny previously approved these changes Sep 8, 2023

View reviewed changes

infiloop2 previously approved these changes Sep 8, 2023

View reviewed changes

Re run CI

366ee69

Comment flake

9a59647

ferglor dismissed stale reviews from infiloop2 and EasterTheBunny via 9a59647 September 8, 2023 19:07

Side step flake

3edaded

FelixFan1992 approved these changes Sep 8, 2023

View reviewed changes

ferglor added this pull request to the merge queue Sep 8, 2023

github-merge-queue bot removed this pull request from the merge queue due to failed status checks Sep 8, 2023

infiloop2 added this pull request to the merge queue Sep 8, 2023

github-merge-queue bot removed this pull request from the merge queue due to failed status checks Sep 8, 2023

infiloop2 added this pull request to the merge queue Sep 8, 2023

Merged via the queue into develop with commit dc6e52b Sep 8, 2023
117 checks passed

infiloop2 deleted the feature/AUTO-4895 branch September 8, 2023 22:29

snehaagni mentioned this pull request Oct 18, 2023

release/2.6.0 -> master #10997

Merged

		copy(cloneRecords, u.pendingRecords)
		u.pendingRecords = []persistedStateRecord{}

		cloneRecords := u.pendingRecords[:batch]
		u.pendingRecords = u.pendingRecords[batch:]

Batch insert records into the upkeep state store #10488

Batch insert records into the upkeep state store #10488

Conversation

ferglor commented Sep 5, 2023

github-actions bot commented Sep 5, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

infiloop2 left a comment

Choose a reason for hiding this comment

github-actions bot commented Sep 7, 2023

github-actions bot commented Sep 7, 2023

github-actions bot commented Sep 7, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Sep 8, 2023

Choose a reason for hiding this comment

infiloop2 Sep 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

infiloop2 Sep 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Sep 8, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

infiloop2 left a comment

Choose a reason for hiding this comment

github-actions bot commented Sep 8, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Sep 8, 2023

EasterTheBunny left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Sep 8, 2023

github-actions bot commented Sep 8, 2023

github-actions bot commented Sep 8, 2023

github-actions bot commented Sep 8, 2023

cl-sonarqube-production bot commented Sep 8, 2023

infiloop2 Sep 8, 2023 •

edited

Loading

infiloop2 Sep 8, 2023 •

edited

Loading