plugins: Fix race between manager and plugin startup

This change fixes a race condition in the manager that was caused by
registering the storage trigger _after_ the plugins had been
started. The problem was that if the bundle plugin was able to
download and activate before the trigger registration in the manager
went through, the store and the manager would be out-of-sync after
startup. The bundle would activate successfully but the plugin
manager would not see the change. This meant that the server health
check, status plugin, etc. would report successful activation and
clients using either of those APIs for synchronization could start
querying. If they executed a query within this window, virtual docs
would not be visible because the plugin manager would not yet have a
compiler to return to the server. Similarly, if clients queried the
v1/policies API they would see the raw policy contents but no AST
(since the latter is retrieved from the compiler.)

To remove the race condition the plugin manager simply registers the
trigger before starting any of the plugins. This ensures that it sees
all changes made by any of the plugins.

Fixes open-policy-agent#2343

Signed-off-by: Torin Sandall <torinsandall@gmail.com>

plugins/plugins.go

@@ -261,20 +261,24 @@ func (m *Manager) RegisterCompilerTrigger(f func(txn storage.Transaction)) {
 
 // Start starts the manager.
 func (m *Manager) Start(ctx context.Context) error {
+
 	if m == nil {
 		return nil
 	}
 
-	err := storage.Txn(ctx, m.Store, storage.TransactionParams{}, func(txn storage.Transaction) error {
-		compiler, err := loadCompilerFromStore(ctx, m.Store, txn)
+	if err := storage.Txn(ctx, m.Store, storage.WriteParams, func(txn storage.Transaction) error {
+
+		c, err := loadCompilerFromStore(ctx, m.Store, txn)
 		if err != nil {
 			return err
 		}
-		m.setCompiler(compiler)
-		return nil
-	})
 
-	if err != nil {
+		m.setCompiler(c)
+
+		_, err = m.Store.Register(ctx, txn, storage.TriggerConfig{OnCommit: m.onCommit})
+		return err
+
+	}); err != nil {
 		return err
 	}
 
@@ -295,12 +299,7 @@ func (m *Manager) Start(ctx context.Context) error {
 		}
 	}
 
-	config := storage.TriggerConfig{OnCommit: m.onCommit}
-
-	return storage.Txn(ctx, m.Store, storage.WriteParams, func(txn storage.Transaction) error {
-		_, err := m.Store.Register(ctx, txn, config)
-		return err
-	})
+	return nil
 }
 
 // Stop stops the manager, stopping all the plugins registered with it
@@ -402,6 +401,7 @@ func (m *Manager) copyPluginStatus() map[string]*Status {
 }
 
 func (m *Manager) onCommit(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
+
 	if event.PolicyChanged() {
 
 		var compiler *ast.Compiler

runtime/runtime.go

@@ -192,6 +192,8 @@ func NewRuntime(ctx context.Context, params Params) (*Runtime, error) {
 		return nil, errors.Wrap(err, "load error")
 	}
 
+	// TOOD(tsandall): All of this storage setup could be done by the plugin manager.
+	// This would avoid the need to parse and recompile modules provided on startup.
 	store := inmem.New()
 
 	txn, err := store.NewTransaction(ctx, storage.WriteParams)