Skip to content

Update aggregator batch processing to maintain in-memory pin state until OnFinalize #483

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

Merged
merged 9 commits into from
Feb 7, 2022
Merged

Update aggregator batch processing to maintain in-memory pin state until OnFinalize #483

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
72e2ede
Update aggregator batch processing to maintain in-memory pin state un…
peterbroadhurst Feb 6, 2022
2be9661
Round out UT coverage
peterbroadhurst Feb 6, 2022
9e67065
Add batch index and correct filters on pins
peterbroadhurst Feb 6, 2022
65bfbfe
Update E2E test to send multiple messages over different topics for b…
peterbroadhurst Feb 6, 2022
651ed24
Merge with main
peterbroadhurst Feb 6, 2022
069547d
Reduce overhead of batch indexing on write
peterbroadhurst Feb 6, 2022
89428af
Merge branch 'main' of github.com:hyperledger/firefly into fix-481
peterbroadhurst Feb 7, 2022
0223291
Add commentary and change scope of NextPinState
peterbroadhurst Feb 7, 2022
8b73337
Tweaks to tests
peterbroadhurst Feb 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions db/migrations/postgres/000059_create_pins_batch_index.down.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
BEGIN;
DROP INDEX pins_batch;
COMMIT;
3 changes: 3 additions & 0 deletions db/migrations/postgres/000059_create_pins_batch_index.up.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
BEGIN;
CREATE INDEX pins_batch ON pins(batch_id);
COMMIT;
1 change: 1 addition & 0 deletions db/migrations/sqlite/000059_create_pins_batch_index.down.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
DROP INDEX pins_batch;
1 change: 1 addition & 0 deletions db/migrations/sqlite/000059_create_pins_batch_index.up.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
CREATE INDEX pins_batch ON pins(batch_id);
25 changes: 13 additions & 12 deletions internal/database/sqlcommon/pin_sql.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright © 2021 Kaleido, Inc.
// Copyright © 2022 Kaleido, Inc.
//
// SPDX-License-Identifier: Apache-2.0
//
Expand Down Expand Up @@ -141,24 +141,25 @@ func (s *SQLCommon) GetPins(ctx context.Context, filter database.Filter) (messag

}

func (s *SQLCommon) SetPinDispatched(ctx context.Context, sequence int64) (err error) {
func (s *SQLCommon) UpdatePins(ctx context.Context, filter database.Filter, update database.Update) (err error) {

ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
if err != nil {
return err
}
defer s.rollbackTx(ctx, tx, autoCommit)

_, err = s.updateTx(ctx, tx, sq.
Update("pins").
Set("dispatched", true).
Where(sq.Eq{
sequenceColumn: sequence,
}),
func() {
s.callbacks.OrderedCollectionEvent(database.CollectionPins, fftypes.ChangeEventTypeUpdated, sequence)
},
)
query, err := s.buildUpdate(sq.Update("pins"), update, pinFilterFieldMap)
if err != nil {
return err
}

query, err = s.filterUpdate(ctx, "", query, filter, pinFilterFieldMap)
if err != nil {
return err
}

_, err = s.updateTx(ctx, tx, query, nil /* no change events filter based update */)
Copy link
Contributor

@awrichar awrichar Feb 7, 2022

Choose a reason for hiding this comment

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

So this type does reliably emit ChangeEventTypeCreated. It theoretically emits ChangeEventTypeDeleted, although I don't know of anywhere we call DeletePin. It now never emits ChangeEventTypeUpdated though.

I wonder if having change events at all is useful, or if it's confusing to emit them inconsistently.

Copy link
Contributor Author

@peterbroadhurst peterbroadhurst Feb 7, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

This is common across the codebase for these update-multiple style actions.

We don't have a need for the events where we use them currently, and it's hard in SQL without detailed DB-specific coding to find out what rows have been updated. There's also a performance overhead

if err != nil {
return err
}
Expand Down
31 changes: 25 additions & 6 deletions internal/database/sqlcommon/pin_sql_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@ func TestPinsE2EWithDB(t *testing.T) {
}

s.callbacks.On("OrderedCollectionEvent", database.CollectionPins, fftypes.ChangeEventTypeCreated, mock.Anything).Return()
s.callbacks.On("OrderedCollectionEvent", database.CollectionPins, fftypes.ChangeEventTypeUpdated, mock.Anything).Return()
s.callbacks.On("OrderedCollectionEvent", database.CollectionPins, fftypes.ChangeEventTypeDeleted, mock.Anything).Return()

err := s.UpsertPin(ctx, pin)
Expand All @@ -67,7 +66,7 @@ func TestPinsE2EWithDB(t *testing.T) {
assert.Equal(t, int64(1), *res.TotalCount)

// Set it dispatched
err = s.SetPinDispatched(ctx, pin.Sequence)
err = s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", pin.Sequence), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
assert.NoError(t, err)

// Double insert, checking no error and we keep the dispatched flag
Expand Down Expand Up @@ -166,22 +165,42 @@ func TestGetPinReadMessageFail(t *testing.T) {
assert.NoError(t, mock.ExpectationsWereMet())
}

func TestSetPinsDispatchedBeginFail(t *testing.T) {
func TestUpdatePinsBeginFail(t *testing.T) {
s, mock := newMockProvider().init()
mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
err := s.SetPinDispatched(context.Background(), 12345)
ctx := context.Background()
err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
assert.Regexp(t, "FF10114", err)
}

func TestSetPinsDispatchedUpdateFail(t *testing.T) {
func TestUpdatePinsUpdateFail(t *testing.T) {
s, mock := newMockProvider().init()
mock.ExpectBegin()
mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
mock.ExpectRollback()
err := s.SetPinDispatched(context.Background(), 12345)
ctx := context.Background()
err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
assert.Regexp(t, "FF10117", err)
}

func TestUpdatePinsBadFilter(t *testing.T) {
s, mock := newMockProvider().init()
mock.ExpectBegin()
mock.ExpectRollback()
ctx := context.Background()
err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("bad", true))
assert.Regexp(t, "FF10148", err)
}

func TestUpdatePinsBadUpdate(t *testing.T) {
s, mock := newMockProvider().init()
mock.ExpectBegin()
mock.ExpectRollback()
ctx := context.Background()
err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("bad", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
assert.Regexp(t, "FF10148", err)
}

func TestPinDeleteBeginFail(t *testing.T) {
s, mock := newMockProvider().init()
mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
Expand Down
Loading