refactor: consolidate oracle and provider orchestrator

cmd/slinky/main.go

@@ -21,7 +21,6 @@
 
 	"github.com/skip-mev/slinky/cmd/build"
 	oraclemetrics "github.com/skip-mev/slinky/oracle/metrics"
-	"github.com/skip-mev/slinky/oracle/orchestrator"
 	"github.com/skip-mev/slinky/pkg/log"
 	oraclemath "github.com/skip-mev/slinky/pkg/math/oracle"
 	oraclefactory "github.com/skip-mev/slinky/providers/factories/oracle"
@@ -254,50 +253,35 @@
 		return fmt.Errorf("failed to create data aggregator: %w", err)
 	}
 
-	// Define the orchestrator and oracle options. These determine how the orchestrator and oracle are created & executed.
-	orchestratorOpts := []orchestrator.Option{
-		orchestrator.WithLogger(logger),
-		orchestrator.WithMarketMap(marketCfg),
-		orchestrator.WithPriceAPIQueryHandlerFactory(oraclefactory.APIQueryHandlerFactory),             // Replace with custom API query handler factory.
-		orchestrator.WithPriceWebSocketQueryHandlerFactory(oraclefactory.WebSocketQueryHandlerFactory), // Replace with custom websocket query handler factory.
-		orchestrator.WithMarketMapperFactory(oraclefactory.MarketMapProviderFactory),
-		orchestrator.WithAggregator(aggregator),
-	}
-	if updateMarketCfgPath != "" {
-		orchestratorOpts = append(orchestratorOpts, orchestrator.WithWriteTo(updateMarketCfgPath))
-	}
+	// Define the oracle options. These determine how the oracle is created & executed.
 	oracleOpts := []oracle.Option{
 		oracle.WithLogger(logger),
-		oracle.WithUpdateInterval(cfg.UpdateInterval),
+		oracle.WithMarketMap(marketCfg),
+		oracle.WithPriceAPIQueryHandlerFactory(oraclefactory.APIQueryHandlerFactory),             // Replace with custom API query handler factory.
+		oracle.WithPriceWebSocketQueryHandlerFactory(oraclefactory.WebSocketQueryHandlerFactory), // Replace with custom websocket query handler factory.
+		oracle.WithMarketMapperFactory(oraclefactory.MarketMapProviderFactory),
 		oracle.WithMetrics(metrics),
-		oracle.WithMaxCacheAge(cfg.MaxPriceAge),
-		oracle.WithPriceAggregator(aggregator),
-	}
-
-	// Create the orchestrator and start the orchestrator.
-	orch, err := orchestrator.NewProviderOrchestrator(
-		cfg,
-		orchestratorOpts...,
-	)
-	if err != nil {
-		return fmt.Errorf("failed to create provider orchestrator: %w", err)
 	}
-
-	if err := orch.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start provider orchestrator: %w", err)
+	if updateMarketCfgPath != "" {
+		oracleOpts = append(oracleOpts, oracle.WithWriteTo(updateMarketCfgPath))
 	}
-	defer orch.Stop()
 
-	// Create the oracle and start the oracle server.
-	oracleOpts = append(
-		oracleOpts,
-		oracle.WithProviders(orch.GetPriceProviders()),
-		oracle.WithMarketMapGetter(orch.GetMarketMap),
+	// Create the oracle and start the oracle.
+	orc, err := oracle.New(
+		cfg,
+		aggregator,
+		oracleOpts...,
 	)
-	orc, err := oracle.New(oracleOpts...)
 	if err != nil {
 		return fmt.Errorf("failed to create oracle: %w", err)
 	}
+	go func() {
+		if err := orc.Start(ctx); err != nil {
+			logger.Fatal("failed to start oracle", zap.Error(err))
+		}
+	}()
+	defer orc.Stop()
+
 	srv := oracleserver.NewOracleServer(orc, logger)
 
 	// cancel oracle on interrupt or terminate
@@ -337,7 +321,7 @@
 		}()
 	}
 
-	// start oracle + server, and wait for either to finish
+	// start server (blocks).
 	if err := srv.StartServer(ctx, cfg.Host, cfg.Port); err != nil {
 		logger.Error("stopping server", zap.Error(err))
 	}

oracle/README.md

@@ -0,0 +1,26 @@
+# Oracle
+
+Responsibilities of the Oracle:
+
+* Spinning up the necessary services (e.g. the sidecar, the price fetchers, etc.).
+* Managing the lifecycle of the providers.
+* Determining the set of markets that need to be fetched and updating the providers accordingly.
+
+## Configuration
+
+At a high level the oracle is configured with a `oracle.json` file that contains all providers that need to be instantiated. To read more about the configuration of `oracle.json`, please refer to the [oracle configuration documentation](config/README.md).
+
+Each provider is instantiated using the `PriceAPIQueryHandlerFactory`, `PriceWebSocketQueryHandlerFactory`, and `MarketMapFactory` factory functions. Think of these as the constructors for the providers. 
+
+* `PriceAPIQueryHandlerFactory` - This is used to create the API query handler for the provider - which is then passed into a base provider.
+* `PriceWebSocketQueryHandlerFactory` - This is used to create the WebSocket query handler for the provider - which is then passed into a base provider.
+* `MarketMapFactory` - This is used to create the market map provider.
+
+## Lifecycle
+
+The oracle can be initialized with an option of `WithMarketMap` which allows each provider to be instantiated with a predetermined set of markets. If this option is not provided, the oracle will fetch the markets from the market map provider. **Both options can be set.**
+
+The oracle will then start each provider in a separate goroutine. Additionally, if the oracle has a market map provider, it will start a goroutine that will periodically fetch the markets from the market map provider and update the providers accordingly.
+
+All providers are running concurrently and will do so until the main context is canceled (what is passed into `Start`). If the oracle is canceled, it will cancel all providers and wait for them to finish before returning.
+

oracle/orchestrator/helpers_test.go → oracle/helpers_test.go

@@ -1,4 +1,4 @@
-package orchestrator_test
+package oracle_test
 
 import (
 	"testing"
@@ -8,8 +8,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/skip-mev/slinky/oracle"
 	"github.com/skip-mev/slinky/oracle/config"
-	"github.com/skip-mev/slinky/oracle/orchestrator"
 	oracletypes "github.com/skip-mev/slinky/oracle/types"
 	slinkytypes "github.com/skip-mev/slinky/pkg/types"
 	"github.com/skip-mev/slinky/providers/apis/binance"
@@ -196,14 +196,33 @@ var (
 		},
 	}
 )
+var _ oracle.PriceAggregator = &noOpPriceAggregator{}
+
+type noOpPriceAggregator struct{}
+
+func (n noOpPriceAggregator) SetProviderPrices(_ string, _ oracletypes.Prices) {
+}
+
+func (n noOpPriceAggregator) UpdateMarketMap(_ mmtypes.MarketMap) {
+}
+
+func (n noOpPriceAggregator) AggregatePrices() {
+}
+
+func (n noOpPriceAggregator) GetPrices() oracletypes.Prices {
+	return oracletypes.Prices{}
+}
+
+func (n noOpPriceAggregator) Reset() {
+}
 
 func checkProviderState(
 	t *testing.T,
 	expectedTickers []oracletypes.ProviderTicker,
 	expectedName string,
 	expectedType providertypes.ProviderType,
 	isRunning bool,
-	state orchestrator.ProviderState,
+	state oracle.ProviderState,
 ) {
 	t.Helper()
 

oracle/orchestrator/init.go → oracle/init.go

@@ -1,4 +1,4 @@
-package orchestrator
+package oracle
 
 import (
 	"context"
@@ -14,7 +14,7 @@ import (
 )
 
 // Init initializes the all providers that are configured via the oracle config.
-func (o *ProviderOrchestrator) Init(ctx context.Context) error {
+func (o *OracleImpl) Init(ctx context.Context) error {
 	o.mut.Lock()
 	defer o.mut.Unlock()
 
@@ -45,7 +45,7 @@ func (o *ProviderOrchestrator) Init(ctx context.Context) error {
 }
 
 // createPriceProvider creates a new price provider for the given provider configuration.
-func (o *ProviderOrchestrator) createPriceProvider(ctx context.Context, cfg config.ProviderConfig) error {
+func (o *OracleImpl) createPriceProvider(ctx context.Context, cfg config.ProviderConfig) error {
 	// Create the provider market map. This creates the tickers the provider is configured to
 	// support.
 	tickers, err := types.ProviderTickersFromMarketMap(cfg.Name, o.marketMap)
@@ -99,8 +99,8 @@ func (o *ProviderOrchestrator) createPriceProvider(ctx context.Context, cfg conf
 		Cfg:      cfg,
 	}
 
-	// Add the provider to the orchestrator.
-	o.providers[provider.Name()] = state
+	// Add the provider to the oracle.
+	o.priceProviders[provider.Name()] = state
 
 	o.logger.Info(
 		"created price provider state",
@@ -111,7 +111,7 @@ func (o *ProviderOrchestrator) createPriceProvider(ctx context.Context, cfg conf
 }
 
 // createAPIQueryHandler creates a new API query handler for the given provider configuration.
-func (o *ProviderOrchestrator) createAPIQueryHandler(
+func (o *OracleImpl) createAPIQueryHandler(
 	ctx context.Context,
 	cfg config.ProviderConfig,
 ) (types.PriceAPIQueryHandler, error) {
@@ -123,7 +123,7 @@ func (o *ProviderOrchestrator) createAPIQueryHandler(
 }
 
 // createWebSocketQueryHandler creates a new web socket query handler for the given provider configuration.
-func (o *ProviderOrchestrator) createWebSocketQueryHandler(
+func (o *OracleImpl) createWebSocketQueryHandler(
 	ctx context.Context,
 	cfg config.ProviderConfig,
 ) (types.PriceWebSocketQueryHandler, error) {
@@ -135,7 +135,7 @@ func (o *ProviderOrchestrator) createWebSocketQueryHandler(
 }
 
 // createMarketMapProvider creates a new market map provider for the given provider configuration.
-func (o *ProviderOrchestrator) createMarketMapProvider(cfg config.ProviderConfig) error {
+func (o *OracleImpl) createMarketMapProvider(cfg config.ProviderConfig) error {
 	if o.marketMapperFactory == nil {
 		return fmt.Errorf("cannot create market map provider; market map factory is not set")
 	}