Fix OPA deadlock while stopping bundle plugin

This commit fixes couple of issues that could result
in blocking OPA:

1) When the bundle plugin attempts to stop the bundle
downloader, it first grabs the lock on the plugin and then
stops the downloader. The downloader in-turn calls the plugin’s
callback function which now waits for the lock to be released
by the plugin's stop function. This results in a deadlock.

This commit fixes this issue by making sure the plugin's stop
function releases the lock before stopping the downloader.

2) Another issue that could block OPA is when the stop function
on the same downloader gets called multiple times.

Fixes: #3363

Signed-off-by: Ashutosh Narkar <anarkar4387@gmail.com>

download/download.go

@@ -54,6 +54,8 @@ type Downloader struct {
 	respHdrTimeoutSec int64
 	wg                sync.WaitGroup
 	logger            logging.Logger
+	mtx               sync.Mutex
+	stopped           bool
 }
 
 // New returns a new Downloader that can be started.
@@ -112,11 +114,19 @@ func (d *Downloader) doStart(context.Context) {
 	done := <-d.stop // blocks until there's something to read
 	cancel()
 	d.wg.Wait()
+	d.stopped = true
 	close(done)
 }
 
 // Stop tells the Downloader to stop downloading bundles.
 func (d *Downloader) Stop(context.Context) {
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
+
+	if d.stopped {
+		return
+	}
+
 	done := make(chan struct{})
 	d.stop <- done
 	<-done

download/download_test.go

@@ -56,6 +56,46 @@ func TestStartStop(t *testing.T) {
 	d.Stop(ctx)
 }
 
+func TestStopWithMultipleCalls(t *testing.T) {
+	ctx := context.Background()
+	fixture := newTestFixture(t)
+
+	updates := make(chan *Update)
+
+	config := Config{}
+	if err := config.ValidateAndInjectDefaults(); err != nil {
+		t.Fatal(err)
+	}
+
+	d := New(config, fixture.client, "/bundles/test/bundle1").WithCallback(func(_ context.Context, u Update) {
+		updates <- &u
+	})
+
+	d.Start(ctx)
+
+	// Give time for some download events to occur
+	time.Sleep(1 * time.Second)
+
+	u1 := <-updates
+
+	if u1.Bundle == nil || len(u1.Bundle.Modules) == 0 {
+		t.Fatal("expected bundle with at least one module but got:", u1)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		d.Stop(ctx)
+		close(done)
+	}()
+
+	d.Stop(ctx)
+	<-done
+
+	if !d.stopped {
+		t.Fatal("expected downloader to be stopped")
+	}
+}
+
 func TestStartStopWithLongPollNotSupported(t *testing.T) {
 	ctx := context.Background()
 

plugins/bundle/plugin.go

@@ -42,6 +42,7 @@ type Plugin struct {
 	cfgMtx            sync.Mutex
 	ready             bool
 	bundlePersistPath string
+	stopped           bool
 }
 
 // New returns a new Plugin with the given config.
@@ -99,7 +100,6 @@ func (p *Plugin) Start(ctx context.Context) error {
 
 	p.initDownloaders()
 	for name, dl := range p.downloaders {
-
 		p.log(name).Info("Starting bundle loader.")
 		dl.Start(ctx)
 	}
@@ -109,8 +109,15 @@ func (p *Plugin) Start(ctx context.Context) error {
 // Stop stops the plugin.
 func (p *Plugin) Stop(ctx context.Context) {
 	p.mtx.Lock()
-	defer p.mtx.Unlock()
+	stopDownloaders := map[string]bundleLoader{}
 	for name, dl := range p.downloaders {
+		stopDownloaders[name] = dl
+	}
+	p.downloaders = nil
+	p.stopped = true
+	p.mtx.Unlock()
+
+	for name, dl := range stopDownloaders {
 		p.log(name).Info("Stopping bundle loader.")
 		dl.Stop(ctx)
 	}
@@ -361,7 +368,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) {
 	if u.Error != nil {
 		p.log(name).Error("Bundle load failed: %v", u.Error)
 		p.status[name].SetError(u.Error)
-		p.downloaders[name].ClearCache()
+		if !p.stopped {
+			p.downloaders[name].ClearCache()
+		}
 		return
 	}
 
@@ -376,7 +385,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) {
 		if err := p.activate(ctx, name, u.Bundle); err != nil {
 			p.log(name).Error("Bundle activation failed: %v", err)
 			p.status[name].SetError(err)
-			p.downloaders[name].ClearCache()
+			if !p.stopped {
+				p.downloaders[name].ClearCache()
+			}
 			return
 		}
 
@@ -387,7 +398,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) {
 			if err != nil {
 				p.log(name).Error("Persisting bundle to disk failed: %v", err)
 				p.status[name].SetError(err)
-				p.downloaders[name].ClearCache()
+				if !p.stopped {
+					p.downloaders[name].ClearCache()
+				}
 				return
 			}
 			p.log(name).Debug("Bundle persisted to disk successfully at path %v.", filepath.Join(p.bundlePersistPath, name))

plugins/bundle/plugin_test.go

@@ -107,6 +107,70 @@ func TestPluginStart(t *testing.T) {
 	}
 }
 
+func TestStop(t *testing.T) {
+	var longPollTimeout int64 = 3
+	done := make(chan struct{})
+	tsURLBase := "/opa-test/"
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.HasPrefix(r.URL.Path, tsURLBase) {
+			t.Fatalf("Invalid request URL path: %s, expected prefix %s", r.URL.Path, tsURLBase)
+		}
+
+		close(done)
+
+		// simulate long operation
+		time.Sleep(time.Duration(longPollTimeout) * time.Second)
+		fmt.Fprintln(w) // Note: this is an invalid bundle and will fail the download
+	}))
+	defer ts.Close()
+
+	ctx := context.Background()
+	manager := getTestManager()
+
+	serviceName := "test-svc"
+	err := manager.Reconfigure(&config.Config{
+		Services: []byte(fmt.Sprintf("{\"%s\":{ \"url\": \"%s\"}}", serviceName, ts.URL+tsURLBase)),
+	})
+	if err != nil {
+		t.Fatalf("Error configuring plugin manager: %s", err)
+	}
+
+	baseConf := download.Config{Polling: download.PollingConfig{LongPollingTimeoutSeconds: &longPollTimeout}}
+
+	plugin := Plugin{
+		manager:     manager,
+		status:      map[string]*Status{},
+		etags:       map[string]string{},
+		downloaders: map[string]bundleLoader{},
+	}
+	bundleName := "test-bundle"
+	plugin.status[bundleName] = &Status{Name: bundleName}
+
+	callback := func(ctx context.Context, u download.Update) {
+		plugin.oneShot(ctx, bundleName, u)
+	}
+	plugin.downloaders[bundleName] = download.New(baseConf, plugin.manager.Client(serviceName), bundleName).WithCallback(callback)
+
+	err = plugin.Start(ctx)
+	if err != nil {
+		t.Fatal("unexpected error:", err)
+	}
+
+	// Give time for a long poll request to be initiated
+	<-done
+
+	plugin.Stop(ctx)
+
+	if plugin.status[bundleName].Code != errCode {
+		t.Fatalf("expected error code %v but got %v", errCode, plugin.status[bundleName].Code)
+	}
+
+	if !strings.Contains(plugin.status[bundleName].Message, "context canceled") {
+		t.Fatalf("unexpected error message %v", plugin.status[bundleName].Message)
+	}
+}
+
 func TestPluginOneShotBundlePersistence(t *testing.T) {
 
 	ctx := context.Background()