PluginProcessors [CCIP-3147] #98

asoliman92 · 2024-09-03T06:43:38Z

Add PluginProcessor interface to create multiple Processors under OCR plugin.

The idea is to separate the logic of different types of observations and outcomes into separate processors.
This makes it easier to manage and test the logic of each type of observation/outcome without affecting each others.
Some of them will implement state machines (e.g. merkleroot), others might implement simpler logic. (e.g. token)
Also makes running them in parallel more streamlined.
The OCR plugin becomes a coordinator/collector of these SubPlugins.

Example Pseudo code:

	OCRPlugin {
	  nodeID
	  merkleProcessor
	  tokenProcessor
	  feeProcessor
	  ...
	}

	OCRPlugin.Observer {
	  mObs := merkleProcessor.Observer
	  tObs := tokenProcessor.Observer
	  fObs := feeProcessor.Observer
	  return Observation{mObs, tObs, fObs}
	}
	OCRPlugin.Validate {
	  mObs := merkleProcessor.Validate
	  tObs := tokenProcessor.Validate
	  fObs := feeProcessor.Validate
	  check errors for each
	  return nil
	}
	OCRPlugin.Outcome {
	  mOut := merkleProcessor.Outcome
	  tOut := tokenProcessor.Outcome
	  fOut := feeProcessor.Outcome
	  return Outcome{mOut, tOut, fOut}
	}

	OCRPlugin.Report {
		return Report{mOut.X, tOut.Y, fOut.Z}
	}

Notice all PluginProcessor interface functions are using prevOutcome instead of outCtx.
We're interested in the prevOutcome, and it makes it easier to have all decoding on the top level (OCR plugin),
otherwise there might be cyclic dependencies or just complicating the code more.

commit/gasworker/observation.go

commit/observation.go

dimkouv · 2024-09-03T09:26:44Z

mocks/cl-common/chainreader/chain_reader.go

 	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"


Let's keep goimports in sync, so we don't have random changes in PRs

I ran the command to fix imports and it did this :/

commit/outcome.go

dimkouv · 2024-09-03T09:29:24Z

commit/types.go

 }

 // Encode encodes an Outcome deterministically
 func (o Outcome) Encode() ([]byte, error) {

 	// Sort all lists to ensure deterministic serialization
-	o.sort()
+	o.MerkleOutcome.Sort()


I believe both of this are not required anymore, we have deterministic outcomes prior to encoding.

We don't currently sort the outcomes before returning it. We can do so but leaving this here is a good safe guard. WDYT?

Determinism is guaranteed. Encoding should not be mixed with determinism.

Imagine Json Encoding re-ordering your arrays, not nice :D

makramkd

This is a good start but I think we need to improve unit-testability. One idea is to have a generic interface that is implemented by all sub-plugins:

type SubPlugin[QueryType any, ObservationType any, OutcomeType any] interface {
  // Will be needed by merkleroots subplugin
  Query(ctx context.Context, outctx OutcomeContext) (QueryType, error)
  Observation(ctx context.Context, outctx OutcomeContext, query QueryType) (ObservationType, error)
  ValidateObservation(outctx OutcomeContext, query QueryType, ao types.AttributedObservation) error
  Outcome(outctx OutcomeContext, query QueryType, aos []types.AttributedObservation) (OutcomeType, error)
}

Then calls to this can easily be mocked in unit tests, and would help us unit-test the "combiner" functions used in all the OCR phases.

commit/gasworker/observation.go

commit/gasworker/outcome.go

commit/merkleworker/types.go

commit/merkleworker/validate_observation.go

commit/merkleworker/worker.go

commit/observation.go

commit/tokenworker/observation.go

winder · 2024-09-03T11:39:49Z

commit/merkleworker/types.go

+	FChain             map[cciptypes.ChainSelector]int `json:"fChain"`
+}
+
+func (obs MerkleObservation) Encode() ([]byte, error) {


@makramkd @dimkouv what do you think about getting rid of every Encode() function in the plugins? Go allows us easily use reflection to do this in one place rather than sprinkling it throughout the types.

I'm removing the Encode from all subplugins, this was copied while initially moving the existing plugin into merkleroot sub plugin. One Encode at the top level Plugin should be okay IMO.

Sounds good, we also want to replace this with proto soon

winder · 2024-09-03T12:14:45Z

This is a good start but I think we need to improve unit-testability. One idea is to have a generic interface that is implemented by all sub-plugins:
type SubPlugin[QueryType any, ObservationType any, OutcomeType any] interface {
  // Will be needed by merkleroots subplugin
  Query(ctx context.Context, outctx OutcomeContext) (QueryType, error)
  Observation(ctx context.Context, outctx OutcomeContext, query QueryType) (ObservationType, error)
  ValidateObservation(outctx OutcomeContext, query QueryType, ao types.AttributedObservation) error
  Outcome(outctx OutcomeContext, query QueryType, aos []types.AttributedObservation) (OutcomeType, error)
}
Then calls to this can easily be mocked in unit tests, and would help us unit-test the "combiner" functions used in all the OCR phases.

I was thinking about commenting something similar. The problem is that the different observe helpers have different arguments as well, which sort of defeats the purpose of adding this layer of abstraction. IMO you can get the same testability with a simple type alias, i.e.:

type MerkleRootObserver func(ctx context.Context, ranges []plugintypes.ChainRange) []cciptypes.MerkleRootChain

This also makes it so each observer act as a standalone piece of code rather than forcing it to be part of the larger interface.

edit: I may be mistaking the Observer type in merkleworker with the SubPlugin idea. I guess each package would have one SubPlugin implementation.

asoliman92 · 2024-09-03T14:54:12Z

commit/merkleroot/observation.go

-			TokenPrices: p.observer.ObserveTokenPrices(ctx),
-			FChain:      p.observer.ObserveFChain(),
+			MerkleRoots: w.observer.ObserveMerkleRoots(ctx, previousOutcome.RangesSelectedForReport),
+			FChain:      w.observer.ObserveFChain(),


Leaving FChain observed here for now. Will iterate on having Fchain sent to the workers later. to unblock working on tokens and gas in commit plugin.

shared/sub_plugin.go

commit/fees/processor.go

commit/merkleroot/observation.go

asoliman92 · 2024-09-03T17:30:29Z

commit/plugin_e2e_test.go

@@ -9,15 +9,18 @@ import (
 	"time"


Changed current tests to use multiple outcomes (specifically merkleroot.Outcome). No new tests added. Maybe later we'll refactor this to have some mocks and make sure it's aggregating correctly without testing the underlying processors as already each processor will have their own unit tests.

asoliman92 · 2024-09-03T17:35:36Z

shared/chain_support.go

@@ -1,4 +1,4 @@
-package commit
+package shared


Moving it to shared so that we can use without cyclic dependencies. Just changing fields to make them publicly available. No added or updated functionality.

commit/merkleroot/processor.go

commit/plugin.go

commit/report.go

commit/plugin.go

makramkd · 2024-09-04T06:44:24Z

commit/fees/types.go

+func (o Outcome) Sort() {
+	// TODO: implement
+}


I feel like we should just enforce that these outcomes that are produced are always correctly sorted in tests.

Having outcome implement it makes it clear that we need to sort them. I agree we need to test it as well. I'm going to use the sort inside the Outcome function in each of the sub plugins

Don't think I agree with that, test enforcement is always better than this kind of implicit contract ("plz implement")

I don't see the conflict. What's the problem with having both?

commit/plugin.go

makramkd · 2024-09-04T06:47:27Z

shared/chain_support.go

Since this is only used in the commit plugin, wouldn't it be more apt to place it in commit/shared?

I think it can be used later with exec plugin as well. It has nothing to do with commit only logic. Leaving it in shared makes it easier to discover and use if needed in other places.

Would be nice to share this across both plugins 👍

shared/sub_plugin.go

dimkouv

i suggest the following renames

fees -> chainfee
token -> tokenprice

then we get

chainfee.Processor
tokenprice.Processor

dimkouv · 2024-09-04T08:10:19Z

commit/merkleroot/processor.go

+	lggr         logger.Logger
+	observer     Observer
+	readerSyncer *plugincommon.BackgroundReaderSyncer
+	ccipReader   reader.CCIP


I believe it's better if we move the plugin methods in this file.
Then we have:

merkleroot/ processor.go processor_test.go types.go types_test.go

Right now I find it hard to navigate through the different files.

I think it's more readable having different files for observation, outcome, report,..etc. I can change the other processors to follow the same scheme so that navigating files is easier. @makramkd @winder WDYT? In all cases I don't think this needs to be done in this PR. We want to unblock the work on tokens and fees for the time being.

Different strokes for different folks, I like the locality as long as the file is < 1,000 lines.

I like locality if it's < 1000 lines, this one will be above the 1K lines and the tests specifically will be hard to navigate as it will be much more than 1K lines.

I don't have a strong preference, but usually lean towards keeping an interface implementation all in one file. That said execute is huge and I was planning to split it up like this at some point.

Either way I agree that it should be done in a separate PR.

commit/plugin.go

commit/merkleroot/outcome.go

commit/merkleroot/processor.go

commit/merkleroot/types.go

commit/merkleroot/validate_observation.go

winder · 2024-09-04T12:33:00Z

commit/merkleroot/validate_observation.go

+		return nil
+	}
+
+	seenChains := mapset.NewSet[cciptypes.ChainSelector]()


Spent a little time reading about mapset, while I'm still not used to seeing it instead of the stdlib version which would use map[cciptypes.Chainsselector]struct{}, at the very least we can get more performance by using the thread unsafe version.

Suggested change

seenChains := mapset.NewSet[cciptypes.ChainSelector]()

seenChains := mapset.NewThreadUnsafeSet[cciptypes.ChainSelector]()

I'm okay using the NewThreadUnsafeSet. It's just usually very alarming for people when they read it and the performance gain won't be much considering the number of selectors we have. Also won't need to worry if we have any concurrency about it.

winder · 2024-09-04T12:37:32Z

commit/merkleroot/validate_observation.go

+		if !observerSupportedChains.Contains(root.ChainSel) {
+			return fmt.Errorf("found merkle root for chain %d, but this chain is not supported by Observer %d",
+				root.ChainSel, observer)
+		}


Looks like a complete error message can be created by removing this block and adding a new one outside of the loop:

unsupportedChains := observerSupportedChains.Difference(seenChains) if len(unsupportedChains) > 0 { return fmt.Errorf("found merkle root for chains unsupported by oracleID %d: %s", observer, unsupportedChains) }

Added to ticket CCIP-3166 to fix later.

winder · 2024-09-04T12:38:09Z

commit/merkleroot/validate_observation.go

+	return nil
+}
+
+func validateObservedOnRampMaxSeqNums(


Same suggestions as above.

Added to ticket CCIP-3166 to fix later.

winder · 2024-09-04T12:38:20Z

commit/merkleroot/validate_observation.go

+	return nil
+}
+
+func validateObservedOffRampMaxSeqNums(


Same suggestions as above.

Added to ticket CCIP-3166 to fix later.

commit/plugin.go

winder · 2024-09-04T12:48:24Z

commit/plugin.go

+	merkleProcessor   shared.PluginProcessor[merkleroot.Query, merkleroot.Observation, merkleroot.Outcome]
+	tokenProcessor    shared.PluginProcessor[tokenprice.Query, tokenprice.Observation, tokenprice.Outcome]
+	feeProcessor      shared.PluginProcessor[chainfee.Query, chainfee.Observation, chainfee.Outcome]


You're not using these generic types so shared.PluginProcessor should define them as any for a simpler interface.

If you don't need the types, you can now be more concise:

Suggested change

merkleProcessor shared.PluginProcessor[merkleroot.Query, merkleroot.Observation, merkleroot.Outcome]

tokenProcessor shared.PluginProcessor[tokenprice.Query, tokenprice.Observation, tokenprice.Outcome]

feeProcessor shared.PluginProcessor[chainfee.Query, chainfee.Observation, chainfee.Outcome]

merkleProcessor shared.PluginProcessor

tokenProcessor shared.PluginProcessor

feeProcessor shared.PluginProcessor

If you do need the types in some cases, specify the concrete type:

Suggested change

merkleProcessor shared.PluginProcessor[merkleroot.Query, merkleroot.Observation, merkleroot.Outcome]

tokenProcessor shared.PluginProcessor[tokenprice.Query, tokenprice.Observation, tokenprice.Outcome]

feeProcessor shared.PluginProcessor[chainfee.Query, chainfee.Observation, chainfee.Outcome]

merkleProcessor merkleroot.Processor // because we need to call ObserveFChain()'

tokenProcessor shared.PluginProcessor

feeProcessor shared.PluginProcessor

I think we'll need the types when we start testing. and using the plugin makes the one mock enough to use. WDYT?

It's not difficult to cast any to the concrete type if needed in tests. You're not actually using the interface with the current code, so you may as well remove it rather than deal with all the generics.

We've discussed on a call and decided to keep as is for testability and we'll revisit this if we can make it more generic later.

winder · 2024-09-04T12:56:15Z

commit/plugin.go

+	merkleObs, err := p.merkleProcessor.Observation(ctx, prevOutcome.MerkleRootOutcome, merkleroot.Query{})
+	if err != nil {
+		p.lggr.Errorw("failed to get merkle observation", "err", err)
+	}
+	tokenPricesObs, err := p.tokenProcessor.Observation(ctx, prevOutcome.TokenPriceOutcome, tokenprice.Query{})
+	if err != nil {
+		//log error
+		p.lggr.Errorw("failed to get token prices", "err", err)
+	}
+	gasObs, err := p.feeProcessor.Observation(ctx, prevOutcome.ChainFeeOutcome, chainfee.Query{})
+	if err != nil {
+		p.lggr.Errorw("failed to get gas prices", "err", err)
+	}


Except for the deterministic ordering, this could be a fully decoupled generic SubPlugin aggregator.

Replace:

merkleProcessor shared.PluginProcessor tokenProcessor shared.PluginProcessor feeProcessor shared.PluginProcessor

With:

subPlugins map[string]shared.SubPlugin

This loop becomes:

var observations map[string]any for k, processor := range subPlugins { // I suppose args need a little more thought. You could have shared k/v's // Not sure about the query. observations[k] = processor.Observation(ctx, prevOutcome, query) } return Encode(observations)

The neat part here is since you're using polymorphism, the Observation function doesn't change when you add another SubPlugin.

I wanted to do this earlier and didn't go this way to make it simpler, different types will make it more complex once you start having conditions or other mapping for different arguments.

We don't need to do it in this PR, but I think its a natural progression of this work.

winder · 2024-09-04T12:58:17Z

commit/plugin.go

@@ -105,18 +243,18 @@ func (p *Plugin) Close() error {
 	return nil
 }

-func (p *Plugin) decodeOutcome(outcome ocr3types.Outcome) (Outcome, State) {
+func (p *Plugin) decodeOutcome(outcome ocr3types.Outcome) Outcome {


Might as well merge this function with DecodeOutcome now that the special part has been removed.

It's called in multiple places so far. Nicer having it this way instead of copying the logic of empty outcome and the error checking 3 times IMO.

winder · 2024-09-04T13:01:05Z

commit/types.go

+	MerkleObs merkleroot.Observation          `json:"merkleObs"`
+	TokenObs  tokenprice.Observation          `json:"tokenObs"`
+	GasObs    chainfee.Observation            `json:"gasObs"`
+	FChain    map[cciptypes.ChainSelector]int `json:"fChain"`


The next logical step is to have this top level plugin fully generic, I suppose this would end up as map[string]any.

I like the idea

execute/factory.go

winder · 2024-09-04T13:04:48Z

shared/chain_support.go

+func NewCCIPChainSupport(
+	lggr logger.Logger,
+	homeChain reader.HomeChain,
+	oracleIDToP2PID map[commontypes.OracleID]libocrtypes.PeerID,
+	nodeID commontypes.OracleID,
+	destChain cciptypes.ChainSelector,
+) CCIPChainSupport {
+	return CCIPChainSupport{
+		lggr:            lggr,
+		homeChain:       homeChain,
+		oracleIDToP2PID: oracleIDToP2PID,
+		nodeID:          nodeID,
+		destChain:       destChain,
+	}
+}
+


Maybe this thing could be passed into the plugin instead of the home chain? And/or expanded to support all the subconsensus stuff.

Might worth checking the feasability of this 👍

shared/plugin_processor.go

… plugin. The idea is to separate the logic of different types of observations and outcomes into separate processors. This makes it easier to manage and test the logic of each type of observation/outcome without affecting each others. Some of them will implement state machines (e.g. merkleroot), others might implement simpler logic. (e.g. token) Also makes running them in parallel more streamlined. The OCR plugin becomes a coordinator/collector of these SubPlugins. Example Pseudo code: ``` OCRPlugin { nodeID merkleProcessor tokenProcessor feeProcessor ... } OCRPlugin.Observer { mObs := merkleProcessor.Observer tObs := tokenProcessor.Observer fObs := feeProcessor.Observer return Observation{mObs, tObs, fObs} } OCRPlugin.Validate { mObs := merkleProcessor.Validate tObs := tokenProcessor.Validate fObs := feeProcessor.Validate check errors for each return nil } OCRPlugin.Outcome { mOut := merkleProcessor.Outcome tOut := tokenProcessor.Outcome fOut := feeProcessor.Outcome return Outcome{mOut, tOut, fOut} } OCRPlugin.Report { return Report{mOut.X, tOut.Y, fOut.Z} } ``` Notice all PluginProcessor interface functions are using `prevOutcome` instead of `outCtx`. We're interested in the prevOutcome, and it makes it easier to have all decoding on the top level (OCR plugin), otherwise there might be cyclic dependencies or just complicating the code more. Signed-off-by: asoliman <[email protected]> Plugin observations Still outcome is not working Signed-off-by: asoliman <[email protected]> Plugin Outcome and Report Signed-off-by: asoliman <[email protected]> Add tokenworker and some cleanups Signed-off-by: asoliman <[email protected]> Add gasworker Signed-off-by: asoliman <[email protected]> linting Signed-off-by: asoliman <[email protected]> Add subplugin interface Signed-off-by: asoliman <[email protected]> Cleaning up Signed-off-by: asoliman <[email protected]> linting Signed-off-by: asoliman <[email protected]> Add SubPlugin to Mockery Signed-off-by: asoliman <[email protected]> Rename Worker to Processor Signed-off-by: asoliman <[email protected]> Cleaning round Signed-off-by: asoliman <[email protected]> Prallelize sub plugins Signed-off-by: asoliman <[email protected]> Fix mockery Signed-off-by: asoliman <[email protected]> Use SubPlugin interface Signed-off-by: asoliman <[email protected]> linting Signed-off-by: asoliman <[email protected]> run mockery Signed-off-by: asoliman <[email protected]> Add docs Signed-off-by: asoliman <[email protected]> Return empty report when merkleroots not generated Signed-off-by: asoliman <[email protected]> Rename packages Signed-off-by: asoliman <[email protected]> Rename variables Signed-off-by: asoliman <[email protected]> docs and lint Signed-off-by: asoliman <[email protected]> Cleaning Signed-off-by: asoliman <[email protected]> Refactoring based on reviews Signed-off-by: asoliman <[email protected]> Remove encode observation from merkleroot subprocessor Signed-off-by: asoliman <[email protected]> Remove go routines for now to simplify logic Signed-off-by: asoliman <[email protected]> Update commit/plugin.go Co-authored-by: Will Winder <[email protected]> review comments Signed-off-by: asoliman <[email protected]> refactor commit observation to use better names Signed-off-by: asoliman <[email protected]>

github-actions · 2024-09-05T07:14:04Z

Test Coverage

Branch	Coverage
parallel-plugins	66.0%
main	72.6%

asoliman92 requested a review from a team as a code owner September 3, 2024 06:43

asoliman92 marked this pull request as draft September 3, 2024 06:44

dimkouv reviewed Sep 3, 2024

View reviewed changes

makramkd requested changes Sep 3, 2024

View reviewed changes

winder reviewed Sep 3, 2024

View reviewed changes

asoliman92 force-pushed the parallel-plugins branch from 3d37fea to bde95cb Compare September 3, 2024 14:31

asoliman92 commented Sep 3, 2024

View reviewed changes

asoliman92 changed the title ~~Parallel plugins~~ Parallel plugins [CCIP-3147] Sep 3, 2024

asoliman92 requested a review from makramkd September 3, 2024 14:59

asoliman92 commented Sep 3, 2024

View reviewed changes

shared/sub_plugin.go Outdated Show resolved Hide resolved

asoliman92 commented Sep 3, 2024

View reviewed changes

commit/fees/processor.go Outdated Show resolved Hide resolved

asoliman92 commented Sep 3, 2024

View reviewed changes

commit/merkleroot/observation.go Outdated Show resolved Hide resolved

asoliman92 force-pushed the parallel-plugins branch from 29f7101 to 0445c0b Compare September 3, 2024 17:28

asoliman92 commented Sep 3, 2024

View reviewed changes

asoliman92 force-pushed the parallel-plugins branch from 0445c0b to d8e4c12 Compare September 3, 2024 17:36

makramkd reviewed Sep 3, 2024

View reviewed changes

commit/plugin.go Outdated Show resolved Hide resolved

makramkd reviewed Sep 4, 2024

View reviewed changes

asoliman92 force-pushed the parallel-plugins branch from 36f6646 to 0f4b0dd Compare September 4, 2024 07:52

asoliman92 marked this pull request as ready for review September 4, 2024 07:52

asoliman92 requested a review from makramkd September 4, 2024 07:52

dimkouv reviewed Sep 4, 2024

View reviewed changes

commit/plugin.go Outdated Show resolved Hide resolved

dimkouv reviewed Sep 4, 2024

View reviewed changes

commit/plugin.go Outdated Show resolved Hide resolved

dimkouv reviewed Sep 4, 2024

View reviewed changes

commit/plugin.go Outdated Show resolved Hide resolved

dimkouv reviewed Sep 4, 2024

View reviewed changes

commit/merkleroot/outcome.go Outdated Show resolved Hide resolved