-
Notifications
You must be signed in to change notification settings - Fork 1.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[KS-430] Provide an OracleFactory to StandardCapabilities (#14305)
* --wip-- [skip CI] * --wip-- [skip CI] * Add bootstrap peers via capabilities spec * Add peer wrapper to Oracle Factory * Add comment explaining delegate position * Cleanup * Create a key bundle if none exists * Use PR #738 hash * Fix bad merge * Bump chainlink-common * Use in-memory DB for OCR persistance * Also add eth pubkey * Oracle instance spawned * Bump chainlink-common version * Undo some changes * Add changeset + fix interface * Update changeset * Use chainlink-common commit from main branch * Add oracle_factory config to SC spec * Undo a change * Things work as is. Checkpoint. * Update comments. Remove redundant log. * Remove redundant test and bring back log line * Remove redundant JSONConfig * Implement oracle factory transmitter * Stop using pkg/errors * Naming convention * Woops * Fix lint errors * Tidy
- Loading branch information
Showing
21 changed files
with
621 additions
and
39 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| --- | ||
| "chainlink": minor | ||
| --- | ||
|
|
||
| #added oracle support in standard capabilities |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| package generic | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "errors" | ||
| "fmt" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| ocr "github.com/smartcontractkit/libocr/offchainreporting2plus" | ||
| "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" | ||
|
|
||
| "github.com/smartcontractkit/chainlink/v2/core/logger" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/job" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/keystore/keys/ocr2key" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/ocrcommon" | ||
| "github.com/smartcontractkit/chainlink/v2/core/services/telemetry" | ||
|
|
||
| "github.com/smartcontractkit/chainlink-common/pkg/types" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/types/core" | ||
| ) | ||
|
|
||
| type oracleFactory struct { | ||
| database ocr3types.Database | ||
| jobID int32 | ||
| jobName string | ||
| jobORM job.ORM | ||
| kb ocr2key.KeyBundle | ||
| lggr logger.Logger | ||
| config job.OracleFactoryConfig | ||
| peerWrapper *ocrcommon.SingletonPeerWrapper | ||
| relayerSet *RelayerSet | ||
| transmitterID string | ||
| } | ||
|
|
||
| type OracleFactoryParams struct { | ||
| JobID int32 | ||
| JobName string | ||
| JobORM job.ORM | ||
| KB ocr2key.KeyBundle | ||
| Logger logger.Logger | ||
| Config job.OracleFactoryConfig | ||
| PeerWrapper *ocrcommon.SingletonPeerWrapper | ||
| RelayerSet *RelayerSet | ||
| TransmitterID string | ||
| } | ||
|
|
||
| func NewOracleFactory(params OracleFactoryParams) (core.OracleFactory, error) { | ||
| return &oracleFactory{ | ||
| database: OracleFactoryDB(params.JobID, params.Logger), | ||
| jobID: params.JobID, | ||
| jobName: params.JobName, | ||
| jobORM: params.JobORM, | ||
| kb: params.KB, | ||
| lggr: params.Logger, | ||
| config: params.Config, | ||
| peerWrapper: params.PeerWrapper, | ||
| relayerSet: params.RelayerSet, | ||
| transmitterID: params.TransmitterID, | ||
| }, nil | ||
| } | ||
|
|
||
| func (of *oracleFactory) NewOracle(ctx context.Context, args core.OracleArgs) (core.Oracle, error) { | ||
| if !of.peerWrapper.IsStarted() { | ||
| return nil, errors.New("peer wrapper not started") | ||
| } | ||
|
|
||
| relayer, err := of.relayerSet.Get(ctx, types.RelayID{Network: of.config.Network, ChainID: of.config.ChainID}) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error when getting relayer: %w", err) | ||
| } | ||
|
|
||
| var relayConfig = struct { | ||
| ChainID string `json:"chainID"` | ||
| EffectiveTransmitterID string `json:"effectiveTransmitterID"` | ||
| SendingKeys []string `json:"sendingKeys"` | ||
| }{ | ||
| ChainID: of.config.ChainID, | ||
| EffectiveTransmitterID: of.transmitterID, | ||
| SendingKeys: []string{of.transmitterID}, | ||
| } | ||
| relayConfigBytes, err := json.Marshal(relayConfig) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error when marshalling relay config: %w", err) | ||
| } | ||
|
|
||
| pluginProvider, err := relayer.NewPluginProvider(ctx, core.RelayArgs{ | ||
| ContractID: of.config.OCRContractAddress, | ||
| ProviderType: "plugin", | ||
| RelayConfig: relayConfigBytes, | ||
| }, core.PluginArgs{ | ||
| TransmitterID: of.transmitterID, | ||
| }) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error when getting offchain digester: %w", err) | ||
| } | ||
|
|
||
| bootstrapPeers, err := ocrcommon.ParseBootstrapPeers(of.config.BootstrapPeers) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to parse bootstrap peers: %w", err) | ||
| } | ||
|
|
||
| oracle, err := ocr.NewOracle(ocr.OCR3OracleArgs[[]byte]{ | ||
| // We are relying on the relayer plugin provider for the offchain config digester | ||
| // and the contract config tracker to save time. | ||
| ContractConfigTracker: pluginProvider.ContractConfigTracker(), | ||
| OffchainConfigDigester: pluginProvider.OffchainConfigDigester(), | ||
| LocalConfig: args.LocalConfig, | ||
| ContractTransmitter: NewContractTransmitter(of.transmitterID, args.ContractTransmitter), | ||
| ReportingPluginFactory: args.ReportingPluginFactoryService, | ||
| BinaryNetworkEndpointFactory: of.peerWrapper.Peer2, | ||
| V2Bootstrappers: bootstrapPeers, | ||
| Database: of.database, | ||
| Logger: ocrcommon.NewOCRWrapper(of.lggr, true, func(ctx context.Context, msg string) { | ||
| logger.Sugared(of.lggr).ErrorIf(of.jobORM.RecordError(ctx, of.jobID, msg), "unable to record error") | ||
| }), | ||
| MonitoringEndpoint: &telemetry.NoopAgent{}, | ||
| OffchainKeyring: of.kb, | ||
| OnchainKeyring: ocrcommon.NewOCR3OnchainKeyringAdapter(of.kb), | ||
| MetricsRegisterer: prometheus.WrapRegistererWith(map[string]string{"job_name": of.jobName}, prometheus.DefaultRegisterer), | ||
| }) | ||
|
|
||
| if err != nil { | ||
| return nil, fmt.Errorf("%w: failed to create new OCR oracle", err) | ||
| } | ||
|
|
||
| return &adaptedOracle{oracle: oracle}, nil | ||
| } | ||
|
|
||
| type adaptedOracle struct { | ||
| oracle ocr.Oracle | ||
| } | ||
|
|
||
| func (a *adaptedOracle) Start(ctx context.Context) error { | ||
| return a.oracle.Start() | ||
| } | ||
|
|
||
| func (a *adaptedOracle) Close(ctx context.Context) error { | ||
| return a.oracle.Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| package generic | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" | ||
|
|
||
| "github.com/smartcontractkit/chainlink/v2/core/logger" | ||
| ) | ||
|
|
||
| type oracleFactoryDb struct { | ||
| // The ID is used for logging and error messages | ||
| // A single standard capabilities spec can instantiate multiple oracles | ||
| // TODO: NewOracle should take a unique identifier for the oracle | ||
| specID int32 | ||
| lggr logger.SugaredLogger | ||
| config *ocrtypes.ContractConfig | ||
| states map[ocrtypes.ConfigDigest]*ocrtypes.PersistentState | ||
| pendingTransmissions map[ocrtypes.ReportTimestamp]ocrtypes.PendingTransmission | ||
| protocolStates map[ocrtypes.ConfigDigest]map[string][]byte | ||
| } | ||
|
|
||
| var ( | ||
| _ ocrtypes.Database = &oracleFactoryDb{} | ||
| ) | ||
|
|
||
| // NewDB returns a new DB scoped to this instanceID | ||
| func OracleFactoryDB(specID int32, lggr logger.Logger) *oracleFactoryDb { | ||
| return &oracleFactoryDb{ | ||
| specID: specID, | ||
| lggr: logger.Sugared(lggr.Named("OracleFactoryMemoryDb")), | ||
| states: make(map[ocrtypes.ConfigDigest]*ocrtypes.PersistentState), | ||
| pendingTransmissions: make(map[ocrtypes.ReportTimestamp]ocrtypes.PendingTransmission), | ||
| protocolStates: make(map[ocrtypes.ConfigDigest]map[string][]byte), | ||
| } | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) ReadState(ctx context.Context, cd ocrtypes.ConfigDigest) (ps *ocrtypes.PersistentState, err error) { | ||
| ps, ok := ofdb.states[cd] | ||
| if !ok { | ||
| return nil, fmt.Errorf("state not found for standard capabilities spec ID %d, config digest %s", ofdb.specID, cd) | ||
| } | ||
|
|
||
| return ps, nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) WriteState(ctx context.Context, cd ocrtypes.ConfigDigest, state ocrtypes.PersistentState) error { | ||
| ofdb.states[cd] = &state | ||
| return nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) ReadConfig(ctx context.Context) (c *ocrtypes.ContractConfig, err error) { | ||
| if ofdb.config == nil { | ||
| // Returning nil, nil because this is a cache miss | ||
| return nil, nil | ||
| } | ||
| return ofdb.config, nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) WriteConfig(ctx context.Context, c ocrtypes.ContractConfig) error { | ||
| ofdb.config = &c | ||
|
|
||
| cBytes, err := json.Marshal(c) | ||
| if err != nil { | ||
| return fmt.Errorf("MemoryDB: WriteConfig failed to marshal config: %w", err) | ||
| } | ||
|
|
||
| ofdb.lggr.Debugw("MemoryDB: WriteConfig", "ocrtypes.ContractConfig", string(cBytes)) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) StorePendingTransmission(ctx context.Context, t ocrtypes.ReportTimestamp, tx ocrtypes.PendingTransmission) error { | ||
| ofdb.pendingTransmissions[t] = tx | ||
| return nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) PendingTransmissionsWithConfigDigest(ctx context.Context, cd ocrtypes.ConfigDigest) (map[ocrtypes.ReportTimestamp]ocrtypes.PendingTransmission, error) { | ||
| m := make(map[ocrtypes.ReportTimestamp]ocrtypes.PendingTransmission) | ||
| for k, v := range ofdb.pendingTransmissions { | ||
| if k.ConfigDigest == cd { | ||
| m[k] = v | ||
| } | ||
| } | ||
|
|
||
| return m, nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) DeletePendingTransmission(ctx context.Context, t ocrtypes.ReportTimestamp) error { | ||
| delete(ofdb.pendingTransmissions, t) | ||
| return nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) DeletePendingTransmissionsOlderThan(ctx context.Context, t time.Time) error { | ||
| for k, v := range ofdb.pendingTransmissions { | ||
| if v.Time.Before(t) { | ||
| delete(ofdb.pendingTransmissions, k) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) ReadProtocolState( | ||
| ctx context.Context, | ||
| configDigest ocrtypes.ConfigDigest, | ||
| key string, | ||
| ) ([]byte, error) { | ||
| value, ok := ofdb.protocolStates[configDigest][key] | ||
| if !ok { | ||
| // Previously implementation returned nil if the state is not found | ||
| return nil, nil | ||
| } | ||
| return value, nil | ||
| } | ||
|
|
||
| func (ofdb *oracleFactoryDb) WriteProtocolState( | ||
| ctx context.Context, | ||
| configDigest ocrtypes.ConfigDigest, | ||
| key string, | ||
| value []byte, | ||
| ) error { | ||
| if value == nil { | ||
| delete(ofdb.protocolStates[configDigest], key) | ||
| } else { | ||
| if ofdb.protocolStates[configDigest] == nil { | ||
| ofdb.protocolStates[configDigest] = make(map[string][]byte) | ||
| } | ||
| ofdb.protocolStates[configDigest][key] = value | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.