From 16b1e575cfe3ae635c362e9861ea3a4f4411a939 Mon Sep 17 00:00:00 2001 From: Matthew Pendrey Date: Wed, 11 Dec 2024 15:23:50 +0000 Subject: [PATCH] lint remove change to try and resolve flaky eth smoke tests --- .mockery.yaml | 6 - .../workflows/syncer/workflow_syncer_test.go | 16 +- .../workflows/syncer/contract_reader_mock.go | 302 ------------------ core/services/workflows/syncer/heap.go | 63 ---- .../workflows/syncer/workflow_registry.go | 90 +----- .../syncer/workflow_registry_test.go | 234 -------------- 6 files changed, 19 insertions(+), 692 deletions(-) delete mode 100644 core/services/workflows/syncer/contract_reader_mock.go delete mode 100644 core/services/workflows/syncer/heap.go delete mode 100644 core/services/workflows/syncer/workflow_registry_test.go diff --git a/.mockery.yaml b/.mockery.yaml index dd9024cc066..5777ca1da92 100644 --- a/.mockery.yaml +++ b/.mockery.yaml @@ -583,12 +583,6 @@ packages: github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer: interfaces: ORM: - ContractReader: - config: - mockname: "Mock{{ .InterfaceName }}" - filename: contract_reader_mock.go - inpackage: true - dir: "{{ .InterfaceDir }}" Handler: config: mockname: "Mock{{ .InterfaceName }}" diff --git a/core/services/relay/evm/capabilities/workflows/syncer/workflow_syncer_test.go b/core/services/relay/evm/capabilities/workflows/syncer/workflow_syncer_test.go index 2ceb59023ec..066e85e839f 100644 --- a/core/services/relay/evm/capabilities/workflows/syncer/workflow_syncer_test.go +++ b/core/services/relay/evm/capabilities/workflows/syncer/workflow_syncer_test.go @@ -108,8 +108,7 @@ func Test_EventHandlerStateSync(t *testing.T) { updateAllowedDONs(t, backendTH, wfRegistryC, []uint32{donID}, true) updateAuthorizedAddress(t, backendTH, wfRegistryC, []common.Address{backendTH.ContractsOwner.From}, true) - // The number of workflows should be greater than the workflow registry contracts pagination limit to ensure - // that the syncer will query the contract multiple times to get the full list of workflows + // Create some initial static state numberWorkflows := 20 for i := 0; i < numberWorkflows; i++ { var workflowID [32]byte @@ -164,11 +163,9 @@ func Test_EventHandlerStateSync(t *testing.T) { testEventHandler.ClearEvents() - // Create events for a number of workflows and confirm that the event handler processes them - + // Create different event types for a number of workflows and confirm that the event handler processes them in order numberOfEventCycles := 50 for i := 0; i < numberOfEventCycles; i++ { - var workflowID [32]byte _, err = rand.Read((workflowID)[:]) require.NoError(t, err) @@ -205,7 +202,6 @@ func Test_EventHandlerStateSync(t *testing.T) { if numEvents == expectedNumEvents { // verify the events are the expected types in the expected order - // Note the below test does not work with the unrefactored workflow registry, event order is essentially random for idx, event := range events { switch idx % 5 { case 0: @@ -566,14 +562,6 @@ func Test_RegistrySyncer_WorkflowRegistered_InitiallyActivated(t *testing.T) { // generate a log event registerWorkflow(t, backendTH, wfRegistryC, giveWorkflow) - /* - string(ForceUpdateSecretsEvent), - string(WorkflowActivatedEvent), - string(WorkflowDeletedEvent), - string(WorkflowPausedEvent), - string(WorkflowRegisteredEvent), - string(WorkflowUpdatedEvent), */ - // Require the secrets contents to eventually be updated require.Eventually(t, func() bool { _, err := er.Get("test-wf") diff --git a/core/services/workflows/syncer/contract_reader_mock.go b/core/services/workflows/syncer/contract_reader_mock.go deleted file mode 100644 index e6e7c8385f5..00000000000 --- a/core/services/workflows/syncer/contract_reader_mock.go +++ /dev/null @@ -1,302 +0,0 @@ -// Code generated by mockery v2.46.3. DO NOT EDIT. - -package syncer - -import ( - context "context" - - query "github.com/smartcontractkit/chainlink-common/pkg/types/query" - primitives "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives" - mock "github.com/stretchr/testify/mock" - - types "github.com/smartcontractkit/chainlink-common/pkg/types" -) - -// MockContractReader is an autogenerated mock type for the ContractReader type -type MockContractReader struct { - mock.Mock -} - -type MockContractReader_Expecter struct { - mock *mock.Mock -} - -func (_m *MockContractReader) EXPECT() *MockContractReader_Expecter { - return &MockContractReader_Expecter{mock: &_m.Mock} -} - -// Bind provides a mock function with given fields: _a0, _a1 -func (_m *MockContractReader) Bind(_a0 context.Context, _a1 []types.BoundContract) error { - ret := _m.Called(_a0, _a1) - - if len(ret) == 0 { - panic("no return value specified for Bind") - } - - var r0 error - if rf, ok := ret.Get(0).(func(context.Context, []types.BoundContract) error); ok { - r0 = rf(_a0, _a1) - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// MockContractReader_Bind_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Bind' -type MockContractReader_Bind_Call struct { - *mock.Call -} - -// Bind is a helper method to define mock.On call -// - _a0 context.Context -// - _a1 []types.BoundContract -func (_e *MockContractReader_Expecter) Bind(_a0 interface{}, _a1 interface{}) *MockContractReader_Bind_Call { - return &MockContractReader_Bind_Call{Call: _e.mock.On("Bind", _a0, _a1)} -} - -func (_c *MockContractReader_Bind_Call) Run(run func(_a0 context.Context, _a1 []types.BoundContract)) *MockContractReader_Bind_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].([]types.BoundContract)) - }) - return _c -} - -func (_c *MockContractReader_Bind_Call) Return(_a0 error) *MockContractReader_Bind_Call { - _c.Call.Return(_a0) - return _c -} - -func (_c *MockContractReader_Bind_Call) RunAndReturn(run func(context.Context, []types.BoundContract) error) *MockContractReader_Bind_Call { - _c.Call.Return(run) - return _c -} - -// Close provides a mock function with given fields: -func (_m *MockContractReader) Close() error { - ret := _m.Called() - - if len(ret) == 0 { - panic("no return value specified for Close") - } - - var r0 error - if rf, ok := ret.Get(0).(func() error); ok { - r0 = rf() - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// MockContractReader_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close' -type MockContractReader_Close_Call struct { - *mock.Call -} - -// Close is a helper method to define mock.On call -func (_e *MockContractReader_Expecter) Close() *MockContractReader_Close_Call { - return &MockContractReader_Close_Call{Call: _e.mock.On("Close")} -} - -func (_c *MockContractReader_Close_Call) Run(run func()) *MockContractReader_Close_Call { - _c.Call.Run(func(args mock.Arguments) { - run() - }) - return _c -} - -func (_c *MockContractReader_Close_Call) Return(_a0 error) *MockContractReader_Close_Call { - _c.Call.Return(_a0) - return _c -} - -func (_c *MockContractReader_Close_Call) RunAndReturn(run func() error) *MockContractReader_Close_Call { - _c.Call.Return(run) - return _c -} - -// GetLatestValueWithHeadData provides a mock function with given fields: ctx, readName, confidenceLevel, params, returnVal -func (_m *MockContractReader) GetLatestValueWithHeadData(ctx context.Context, readName string, confidenceLevel primitives.ConfidenceLevel, params any, returnVal any) (*types.Head, error) { - ret := _m.Called(ctx, readName, confidenceLevel, params, returnVal) - - if len(ret) == 0 { - panic("no return value specified for GetLatestValueWithHeadData") - } - - var r0 *types.Head - var r1 error - if rf, ok := ret.Get(0).(func(context.Context, string, primitives.ConfidenceLevel, any, any) (*types.Head, error)); ok { - return rf(ctx, readName, confidenceLevel, params, returnVal) - } - if rf, ok := ret.Get(0).(func(context.Context, string, primitives.ConfidenceLevel, any, any) *types.Head); ok { - r0 = rf(ctx, readName, confidenceLevel, params, returnVal) - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).(*types.Head) - } - } - - if rf, ok := ret.Get(1).(func(context.Context, string, primitives.ConfidenceLevel, any, any) error); ok { - r1 = rf(ctx, readName, confidenceLevel, params, returnVal) - } else { - r1 = ret.Error(1) - } - - return r0, r1 -} - -// MockContractReader_GetLatestValueWithHeadData_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetLatestValueWithHeadData' -type MockContractReader_GetLatestValueWithHeadData_Call struct { - *mock.Call -} - -// GetLatestValueWithHeadData is a helper method to define mock.On call -// - ctx context.Context -// - readName string -// - confidenceLevel primitives.ConfidenceLevel -// - params any -// - returnVal any -func (_e *MockContractReader_Expecter) GetLatestValueWithHeadData(ctx interface{}, readName interface{}, confidenceLevel interface{}, params interface{}, returnVal interface{}) *MockContractReader_GetLatestValueWithHeadData_Call { - return &MockContractReader_GetLatestValueWithHeadData_Call{Call: _e.mock.On("GetLatestValueWithHeadData", ctx, readName, confidenceLevel, params, returnVal)} -} - -func (_c *MockContractReader_GetLatestValueWithHeadData_Call) Run(run func(ctx context.Context, readName string, confidenceLevel primitives.ConfidenceLevel, params any, returnVal any)) *MockContractReader_GetLatestValueWithHeadData_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(string), args[2].(primitives.ConfidenceLevel), args[3].(any), args[4].(any)) - }) - return _c -} - -func (_c *MockContractReader_GetLatestValueWithHeadData_Call) Return(head *types.Head, err error) *MockContractReader_GetLatestValueWithHeadData_Call { - _c.Call.Return(head, err) - return _c -} - -func (_c *MockContractReader_GetLatestValueWithHeadData_Call) RunAndReturn(run func(context.Context, string, primitives.ConfidenceLevel, any, any) (*types.Head, error)) *MockContractReader_GetLatestValueWithHeadData_Call { - _c.Call.Return(run) - return _c -} - -// QueryKey provides a mock function with given fields: _a0, _a1, _a2, _a3, _a4 -func (_m *MockContractReader) QueryKey(_a0 context.Context, _a1 types.BoundContract, _a2 query.KeyFilter, _a3 query.LimitAndSort, _a4 any) ([]types.Sequence, error) { - ret := _m.Called(_a0, _a1, _a2, _a3, _a4) - - if len(ret) == 0 { - panic("no return value specified for QueryKey") - } - - var r0 []types.Sequence - var r1 error - if rf, ok := ret.Get(0).(func(context.Context, types.BoundContract, query.KeyFilter, query.LimitAndSort, any) ([]types.Sequence, error)); ok { - return rf(_a0, _a1, _a2, _a3, _a4) - } - if rf, ok := ret.Get(0).(func(context.Context, types.BoundContract, query.KeyFilter, query.LimitAndSort, any) []types.Sequence); ok { - r0 = rf(_a0, _a1, _a2, _a3, _a4) - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).([]types.Sequence) - } - } - - if rf, ok := ret.Get(1).(func(context.Context, types.BoundContract, query.KeyFilter, query.LimitAndSort, any) error); ok { - r1 = rf(_a0, _a1, _a2, _a3, _a4) - } else { - r1 = ret.Error(1) - } - - return r0, r1 -} - -// MockContractReader_QueryKey_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'QueryKey' -type MockContractReader_QueryKey_Call struct { - *mock.Call -} - -// QueryKey is a helper method to define mock.On call -// - _a0 context.Context -// - _a1 types.BoundContract -// - _a2 query.KeyFilter -// - _a3 query.LimitAndSort -// - _a4 any -func (_e *MockContractReader_Expecter) QueryKey(_a0 interface{}, _a1 interface{}, _a2 interface{}, _a3 interface{}, _a4 interface{}) *MockContractReader_QueryKey_Call { - return &MockContractReader_QueryKey_Call{Call: _e.mock.On("QueryKey", _a0, _a1, _a2, _a3, _a4)} -} - -func (_c *MockContractReader_QueryKey_Call) Run(run func(_a0 context.Context, _a1 types.BoundContract, _a2 query.KeyFilter, _a3 query.LimitAndSort, _a4 any)) *MockContractReader_QueryKey_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(types.BoundContract), args[2].(query.KeyFilter), args[3].(query.LimitAndSort), args[4].(any)) - }) - return _c -} - -func (_c *MockContractReader_QueryKey_Call) Return(_a0 []types.Sequence, _a1 error) *MockContractReader_QueryKey_Call { - _c.Call.Return(_a0, _a1) - return _c -} - -func (_c *MockContractReader_QueryKey_Call) RunAndReturn(run func(context.Context, types.BoundContract, query.KeyFilter, query.LimitAndSort, any) ([]types.Sequence, error)) *MockContractReader_QueryKey_Call { - _c.Call.Return(run) - return _c -} - -// Start provides a mock function with given fields: ctx -func (_m *MockContractReader) Start(ctx context.Context) error { - ret := _m.Called(ctx) - - if len(ret) == 0 { - panic("no return value specified for Start") - } - - var r0 error - if rf, ok := ret.Get(0).(func(context.Context) error); ok { - r0 = rf(ctx) - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// MockContractReader_Start_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Start' -type MockContractReader_Start_Call struct { - *mock.Call -} - -// Start is a helper method to define mock.On call -// - ctx context.Context -func (_e *MockContractReader_Expecter) Start(ctx interface{}) *MockContractReader_Start_Call { - return &MockContractReader_Start_Call{Call: _e.mock.On("Start", ctx)} -} - -func (_c *MockContractReader_Start_Call) Run(run func(ctx context.Context)) *MockContractReader_Start_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context)) - }) - return _c -} - -func (_c *MockContractReader_Start_Call) Return(_a0 error) *MockContractReader_Start_Call { - _c.Call.Return(_a0) - return _c -} - -func (_c *MockContractReader_Start_Call) RunAndReturn(run func(context.Context) error) *MockContractReader_Start_Call { - _c.Call.Return(run) - return _c -} - -// NewMockContractReader creates a new instance of MockContractReader. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. -// The first argument is typically a *testing.T value. -func NewMockContractReader(t interface { - mock.TestingT - Cleanup(func()) -}) *MockContractReader { - mock := &MockContractReader{} - mock.Mock.Test(t) - - t.Cleanup(func() { mock.AssertExpectations(t) }) - - return mock -} diff --git a/core/services/workflows/syncer/heap.go b/core/services/workflows/syncer/heap.go deleted file mode 100644 index 061293928a3..00000000000 --- a/core/services/workflows/syncer/heap.go +++ /dev/null @@ -1,63 +0,0 @@ -package syncer - -import "container/heap" - -type Heap interface { - // Push adds a new item to the heap. - Push(x WorkflowRegistryEventResponse) - - // Pop removes the smallest item from the heap and returns it. - Pop() WorkflowRegistryEventResponse - - // Len returns the number of items in the heap. - Len() int -} - -// publicHeap is a wrapper around the heap.Interface that exposes the Push and Pop methods. -type publicHeap[T any] struct { - heap heap.Interface -} - -func (h *publicHeap[T]) Push(x T) { - heap.Push(h.heap, x) -} - -func (h *publicHeap[T]) Pop() T { - return heap.Pop(h.heap).(T) -} - -func (h *publicHeap[T]) Len() int { - return h.heap.Len() -} - -// blockHeightHeap is a heap.Interface that sorts WorkflowRegistryEventResponses by block height. -type blockHeightHeap []WorkflowRegistryEventResponse - -// newBlockHeightHeap returns an initialized heap that sorts WorkflowRegistryEventResponses by block height. -func newBlockHeightHeap() Heap { - h := blockHeightHeap(make([]WorkflowRegistryEventResponse, 0)) - heap.Init(&h) - return &publicHeap[WorkflowRegistryEventResponse]{heap: &h} -} - -func (h *blockHeightHeap) Len() int { return len(*h) } - -func (h *blockHeightHeap) Less(i, j int) bool { - return (*h)[i].Event.Head.Height < (*h)[j].Event.Head.Height -} - -func (h *blockHeightHeap) Swap(i, j int) { - (*h)[i], (*h)[j] = (*h)[j], (*h)[i] -} - -func (h *blockHeightHeap) Push(x any) { - *h = append(*h, x.(WorkflowRegistryEventResponse)) -} - -func (h *blockHeightHeap) Pop() any { - old := *h - n := len(old) - x := old[n-1] - *h = old[0 : n-1] - return x -} diff --git a/core/services/workflows/syncer/workflow_registry.go b/core/services/workflows/syncer/workflow_registry.go index 5efb0a9e34e..4db6d26394e 100644 --- a/core/services/workflows/syncer/workflow_registry.go +++ b/core/services/workflows/syncer/workflow_registry.go @@ -91,10 +91,6 @@ type WorkflowLoadConfig struct { // FetcherFunc is an abstraction for fetching the contents stored at a URL. type FetcherFunc func(ctx context.Context, url string) ([]byte, error) -type ContractReaderFactory interface { - NewContractReader(context.Context, []byte) (types.ContractReader, error) -} - // ContractReader is a subset of types.ContractReader defined locally to enable mocking. type ContractReader interface { Start(ctx context.Context) error @@ -104,6 +100,10 @@ type ContractReader interface { GetLatestValueWithHeadData(ctx context.Context, readName string, confidenceLevel primitives.ConfidenceLevel, params any, returnVal any) (head *types.Head, err error) } +type ContractReaderFactory interface { + NewContractReader(context.Context, []byte) (types.ContractReader, error) +} + // WorkflowRegistrySyncer is the public interface of the package. type WorkflowRegistrySyncer interface { services.Service @@ -129,21 +129,11 @@ type workflowRegistry struct { newContractReaderFn newContractReaderFn - eventPollerCfg WorkflowEventPollerConfig - eventTypes []WorkflowRegistryEventType - - // eventsCh is read by the handler and each event is handled once received. - eventsCh chan WorkflowRegistryEventResponse + eventPollerCfg WorkflowEventPollerConfig + eventTypes []WorkflowRegistryEventType handler evtHandler initialWorkflowsStateLoader initialWorkflowsStateLoader - // batchCh is a channel that receives batches of events from the contract query goroutines. - batchCh chan []WorkflowRegistryEventResponse - - // heap is a min heap that merges batches of events from the contract query goroutines. The - // default min heap is sorted by block height. - heap Heap - workflowDonNotifier donNotifier reader ContractReader @@ -198,11 +188,8 @@ func NewWorkflowRegistry( newContractReaderFn: newContractReaderFn, workflowRegistryAddress: addr, eventPollerCfg: eventPollerConfig, - heap: newBlockHeightHeap(), stopCh: make(services.StopChan), eventTypes: ets, - eventsCh: make(chan WorkflowRegistryEventResponse), - batchCh: make(chan []WorkflowRegistryEventResponse, 50000), handler: handler, initialWorkflowsStateLoader: initialWorkflowsStateLoader, workflowDonNotifier: workflowDonNotifier, @@ -245,15 +232,7 @@ func (w *workflowRegistry) Start(_ context.Context) error { return } - w.syncEventsLoop(ctx, reader, loadWorkflowsHead.Height) - }() - - w.wg.Add(1) - go func() { - defer w.wg.Done() - defer cancel() - - w.handlerLoop(ctx) + w.readRegistryEvents(ctx, reader, loadWorkflowsHead.Height) }() return nil @@ -280,37 +259,11 @@ func (w *workflowRegistry) Name() string { return name } -// handlerLoop handles the events that are emitted by the contract. -func (w *workflowRegistry) handlerLoop(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case resp, open := <-w.eventsCh: - if !open { - return - } - - if resp.Err != nil || resp.Event == nil { - w.lggr.Errorw("failed to handle event", "err", resp.Err) - continue - } - - event := resp.Event - w.lggr.Debugf("handling event: %+v", event) - if err := w.handler.Handle(ctx, *event); err != nil { - w.lggr.Errorw("failed to handle event", "event", event, "err", err) - continue - } - } - } -} - -// syncEventsLoop polls the contract for events and passes them to a channel for handling. -func (w *workflowRegistry) syncEventsLoop(ctx context.Context, reader ContractReader, lastReadBlockNumber string) { +// readRegistryEvents polls the contract for events and send them to the events channel. +func (w *workflowRegistry) readRegistryEvents(ctx context.Context, reader ContractReader, lastReadBlockNumber string) { ticker := w.getTicker() - var keyQueries []types.ContractKeyFilter + var keyQueries = make([]types.ContractKeyFilter, 0, len(w.eventTypes)) for _, et := range w.eventTypes { var logData values.Value keyQueries = append(keyQueries, types.ContractKeyFilter{ @@ -356,12 +309,7 @@ func (w *workflowRegistry) syncEventsLoop(ctx context.Context, reader ContractRe EventType: WorkflowRegistryEventType(eventType), }) } - - lcursor := cursor - if lcursor == "" { - lcursor = "empty" - } - w.lggr.Debugw("QueryKeys called", "logs", len(logs), "eventTypes", w.eventTypes, "lastReadBlockNumber", lastReadBlockNumber, "logCursor", lcursor) + w.lggr.Debugw("QueryKeys called", "logs", len(logs), "eventTypes", w.eventTypes, "lastReadBlockNumber", lastReadBlockNumber, "logCursor", cursor) // ChainReader QueryKey API provides logs including the cursor value and not // after the cursor value. If the response only consists of the log corresponding @@ -382,16 +330,12 @@ func (w *workflowRegistry) syncEventsLoop(ctx context.Context, reader ContractRe cursor = log.Sequence.Cursor } - w.sendEvents(ctx, events) - } - } -} - -func (w *workflowRegistry) sendEvents(ctx context.Context, events []WorkflowRegistryEventResponse) { - for _, event := range events { - select { - case <-ctx.Done(): - case w.eventsCh <- event: + for _, event := range events { + err := w.handler.Handle(ctx, event.Event) + if err != nil { + w.lggr.Errorw("failed to handle event", "err", err) + } + } } } } diff --git a/core/services/workflows/syncer/workflow_registry_test.go b/core/services/workflows/syncer/workflow_registry_test.go deleted file mode 100644 index 073de2cbdbc..00000000000 --- a/core/services/workflows/syncer/workflow_registry_test.go +++ /dev/null @@ -1,234 +0,0 @@ -package syncer - -import ( - "context" - "encoding/hex" - "testing" - "time" - - "github.com/stretchr/testify/mock" - - "github.com/jonboulle/clockwork" - - "github.com/smartcontractkit/chainlink-common/pkg/capabilities" - "github.com/smartcontractkit/chainlink-common/pkg/custmsg" - "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" - "github.com/smartcontractkit/chainlink-common/pkg/types" - "github.com/smartcontractkit/chainlink-common/pkg/types/query" - "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives" - "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" - "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest" - "github.com/smartcontractkit/chainlink/v2/core/logger" - "github.com/smartcontractkit/chainlink/v2/core/services/keystore/keys/workflowkey" - "github.com/smartcontractkit/chainlink/v2/core/utils/crypto" - "github.com/smartcontractkit/chainlink/v2/core/utils/matches" - - "github.com/stretchr/testify/require" -) - -type testDonNotifier struct { - don capabilities.DON - err error -} - -func (t *testDonNotifier) WaitForDon(ctx context.Context) (capabilities.DON, error) { - return t.don, t.err -} - -func Test_Workflow_Registry_Syncer(t *testing.T) { - var ( - giveContents = "contents" - wantContents = "updated contents" - contractAddress = "0xdeadbeef" - giveCfg = WorkflowEventPollerConfig{ - QueryCount: 20, - } - giveURL = "http://example.com" - giveHash, err = crypto.Keccak256([]byte(giveURL)) - - giveLog = types.Sequence{ - Data: map[string]any{ - "SecretsURLHash": giveHash, - "Owner": "0xowneraddr", - }, - Cursor: "cursor", - } - ) - - require.NoError(t, err) - - var ( - lggr = logger.TestLogger(t) - db = pgtest.NewSqlxDB(t) - orm = &orm{ds: db, lggr: lggr} - ctx, cancel = context.WithCancel(testutils.Context(t)) - reader = NewMockContractReader(t) - emitter = custmsg.NewLabeler() - gateway = func(_ context.Context, _ string) ([]byte, error) { - return []byte(wantContents), nil - } - ticker = make(chan time.Time) - - handler = NewEventHandler(lggr, orm, gateway, nil, nil, - emitter, clockwork.NewFakeClock(), workflowkey.Key{}) - loader = NewWorkflowRegistryContractLoader(lggr, contractAddress, func(ctx context.Context, bytes []byte) (ContractReader, error) { - return reader, nil - }, handler) - - worker = NewWorkflowRegistry(lggr, func(ctx context.Context, bytes []byte) (ContractReader, error) { - return reader, nil - }, contractAddress, - WorkflowEventPollerConfig{ - QueryCount: 20, - }, handler, loader, - &testDonNotifier{ - don: capabilities.DON{ - ID: 1, - }, - err: nil, - }, - WithTicker(ticker)) - ) - - // Cleanup the worker - defer cancel() - - // Seed the DB with an original entry - _, err = orm.Create(ctx, giveURL, hex.EncodeToString(giveHash), giveContents) - require.NoError(t, err) - - // Mock out the contract reader query - reader.EXPECT().QueryKey( - matches.AnyContext, - types.BoundContract{ - Name: WorkflowRegistryContractName, - Address: contractAddress, - }, - query.KeyFilter{ - Key: string(ForceUpdateSecretsEvent), - Expressions: []query.Expression{ - query.Confidence(primitives.Finalized), - query.Block("0", primitives.Gte), - }, - }, - query.LimitAndSort{ - SortBy: []query.SortBy{query.NewSortByTimestamp(query.Asc)}, - Limit: query.Limit{Count: giveCfg.QueryCount}, - }, - new(values.Value), - ).Return([]types.Sequence{giveLog}, nil) - reader.EXPECT().QueryKey( - matches.AnyContext, - types.BoundContract{ - Name: WorkflowRegistryContractName, - Address: contractAddress, - }, - query.KeyFilter{ - Key: string(WorkflowPausedEvent), - Expressions: []query.Expression{ - query.Confidence(primitives.Finalized), - query.Block("0", primitives.Gte), - }, - }, - query.LimitAndSort{ - SortBy: []query.SortBy{query.NewSortByTimestamp(query.Asc)}, - Limit: query.Limit{Count: giveCfg.QueryCount}, - }, - new(values.Value), - ).Return([]types.Sequence{}, nil) - reader.EXPECT().QueryKey( - matches.AnyContext, - types.BoundContract{ - Name: WorkflowRegistryContractName, - Address: contractAddress, - }, - query.KeyFilter{ - Key: string(WorkflowDeletedEvent), - Expressions: []query.Expression{ - query.Confidence(primitives.Finalized), - query.Block("0", primitives.Gte), - }, - }, - query.LimitAndSort{ - SortBy: []query.SortBy{query.NewSortByTimestamp(query.Asc)}, - Limit: query.Limit{Count: giveCfg.QueryCount}, - }, - new(values.Value), - ).Return([]types.Sequence{}, nil) - reader.EXPECT().QueryKey( - matches.AnyContext, - types.BoundContract{ - Name: WorkflowRegistryContractName, - Address: contractAddress, - }, - query.KeyFilter{ - Key: string(WorkflowActivatedEvent), - Expressions: []query.Expression{ - query.Confidence(primitives.Finalized), - query.Block("0", primitives.Gte), - }, - }, - query.LimitAndSort{ - SortBy: []query.SortBy{query.NewSortByTimestamp(query.Asc)}, - Limit: query.Limit{Count: giveCfg.QueryCount}, - }, - new(values.Value), - ).Return([]types.Sequence{}, nil) - reader.EXPECT().QueryKey( - matches.AnyContext, - types.BoundContract{ - Name: WorkflowRegistryContractName, - Address: contractAddress, - }, - query.KeyFilter{ - Key: string(WorkflowUpdatedEvent), - Expressions: []query.Expression{ - query.Confidence(primitives.Finalized), - query.Block("0", primitives.Gte), - }, - }, - query.LimitAndSort{ - SortBy: []query.SortBy{query.NewSortByTimestamp(query.Asc)}, - Limit: query.Limit{Count: giveCfg.QueryCount}, - }, - new(values.Value), - ).Return([]types.Sequence{}, nil) - reader.EXPECT().QueryKey( - matches.AnyContext, - types.BoundContract{ - Name: WorkflowRegistryContractName, - Address: contractAddress, - }, - query.KeyFilter{ - Key: string(WorkflowRegisteredEvent), - Expressions: []query.Expression{ - query.Confidence(primitives.Finalized), - query.Block("0", primitives.Gte), - }, - }, - query.LimitAndSort{ - SortBy: []query.SortBy{query.NewSortByTimestamp(query.Asc)}, - Limit: query.Limit{Count: giveCfg.QueryCount}, - }, - new(values.Value), - ).Return([]types.Sequence{}, nil) - reader.EXPECT().GetLatestValueWithHeadData(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&types.Head{ - Height: "0", - }, nil) - reader.EXPECT().Start(mock.Anything).Return(nil) - reader.EXPECT().Bind(mock.Anything, mock.Anything).Return(nil) - - // Go run the worker - servicetest.Run(t, worker) - - // Send a tick to start a query - ticker <- time.Now() - - // Require the secrets contents to eventually be updated - require.Eventually(t, func() bool { - secrets, err := orm.GetContents(ctx, giveURL) - require.NoError(t, err) - return secrets == wantContents - }, 5*time.Second, time.Second) -}