Repository: kolide/launcher

Issue:
launcher can get stuck in control server fetch deadlock
Launcher can get stuck with repeatedly logging `fetch is currently executing elsewhere`. We are seeing this issue still on devices running 1.19, so https://github.com/kolide/launcher/pull/2153 does not appear to have fixed the issue.

This issue possibly causes issues with the KATC tables, which require configuration from the control server -- it may be responsible for the `Error executing scheduled query <id>: vtable constructor failed` that we occasionally see.

Discussion:
@RebeccaMahany: The explanation I have for the deadlock:

1. A control request interval override is set
2. At the configured time later, the control request interval override expires; the FlagController calls notifyObservers
3. notifyObservers holds **observersMutex**
4. Since the control request interval has changed, notifyObservers calls ControlService.FlagsChanged
5. ControlService.FlagsChanged calls ControlService.requestIntervalChanged, which calls ControlService.Fetch
6. ControlService.Fetch holds **fetchMutex**
7. ControlService.Fetch receives updated for the KATC subsystem
8. ControlService.Fetch calls subscribers.Ping, which for the KATC subsystem is the osquery runner
9. The KATC config must be rebuilt, which means initializing new tablewrapper tables
10. When creating a tablewrapper table, the constructor calls FlagController.RegisterChangeObserver so the table can respond to changes in TableGenerateTimeout; this call can never finish because FlagController.RegisterChangeObserver cannot acquire **observersMutex**
11. Creating the table never completes, so Fetch never completes

The deadlock is between 3 and 10, and the fetch mutex is held indefinitely by 6 and 11.

Solution patch:
diff --git a/ee/agent/flags/flag_controller.go b/ee/agent/flags/flag_controller.go
index 61ba1957..7f632365 100644
--- a/ee/agent/flags/flag_controller.go
+++ b/ee/agent/flags/flag_controller.go
@@ -135,14 +135,21 @@ func (fc *FlagController) notifyObservers(ctx context.Context, flagKeys ...keys.
 	ctx, span := observability.StartSpan(ctx)
 	defer span.End()
 
+	// If we hold `fc.observersMutex` for the duration of this function, we can have a deadlock
+	// if any `observer.FlagsChanged` function results in calling `fc.RegisterChangeObserver`,
+	// which uses the same `fc.observersMutex`. Therefore, we create a copy of the current
+	// observers map here and release `fc.observersMutex` preemptively.
 	fc.observersMutex.RLock()
 	span.AddEvent("observers_lock_acquired")
-	defer func() {
-		fc.observersMutex.RUnlock()
-		span.AddEvent("observers_lock_released")
-	}()
-
+	currentObservers := make(map[types.FlagsChangeObserver][]keys.FlagKey)
 	for observer, observedKeys := range fc.observers {
+		currentObservers[observer] = make([]keys.FlagKey, len(observedKeys))
+		copy(currentObservers[observer], observedKeys)
+	}
+	fc.observersMutex.RUnlock()
+	span.AddEvent("observers_lock_released")
+
+	for observer, observedKeys := range currentObservers {
 		changedKeys := keys.Intersection(observedKeys, flagKeys)
 
 		if len(changedKeys) > 0 {



Test patch:
diff --git a/ee/agent/flags/flag_controller_test.go b/ee/agent/flags/flag_controller_test.go
index bbf6e3b1..12065ec2 100644
--- a/ee/agent/flags/flag_controller_test.go
+++ b/ee/agent/flags/flag_controller_test.go
@@ -1,9 +1,13 @@
 package flags
 
 import (
+	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"slices"
+
 	"github.com/kolide/launcher/ee/agent/flags/keys"
 	"github.com/kolide/launcher/ee/agent/storage"
 	storageci "github.com/kolide/launcher/ee/agent/storage/ci"
@@ -411,3 +415,82 @@ func TestDeregisterChangeObserver(t *testing.T) {
 		})
 	}
 }
+
+type deadlockedObserver struct {
+	flags                types.Flags
+	observerKey          keys.FlagKey
+	flagsChangedCalled   *atomic.Bool
+	flagsChangedReturned *atomic.Bool
+}
+
+func newDeadlockedObserver(f types.Flags, observerKey keys.FlagKey) *deadlockedObserver {
+	return &deadlockedObserver{
+		flags:                f,
+		observerKey:          observerKey,
+		flagsChangedCalled:   &atomic.Bool{},
+		flagsChangedReturned: &atomic.Bool{},
+	}
+}
+
+func (d *deadlockedObserver) FlagsChanged(ctx context.Context, flagKeys ...keys.FlagKey) {
+	d.flagsChangedCalled.Store(true)
+	// This simulates the control service acquiring the Fetch lock, performing Fetch, getting katc_config updates,
+	// creating a new table with the updated config, and then trying to register that table as a change observer
+	// with the flag controller (so that the table can respond to changes in `d.observerKey` i.e. TableGenerateTimeout).
+	d.flags.RegisterChangeObserver(d, d.observerKey)
+	d.flagsChangedReturned.Store(true)
+}
+
+func TestObserverDeadlock(t *testing.T) {
+	t.Parallel()
+
+	store, err := storageci.NewStore(t, multislogger.NewNopLogger(), storage.AgentFlagsStore.String())
+	require.NoError(t, err)
+	fc := NewFlagController(multislogger.NewNopLogger(), store)
+	assert.NotNil(t, fc)
+
+	// Set up an observer that will observe changes to one key (ControlRequestInterval)
+	// and on change to that interval, will set up a new observer for a different key.
+	newObserverKey := keys.TableGenerateTimeout
+	d := newDeadlockedObserver(fc, newObserverKey)
+	fc.RegisterChangeObserver(d, keys.ControlRequestInterval)
+
+	// Now, set up an override
+	overrideDuration := 30 * time.Second
+	setOverrideReturned := make(chan struct{})
+
+	go func() {
+		fc.SetControlRequestIntervalOverride(3*time.Second, overrideDuration)
+		setOverrideReturned <- struct{}{}
+	}()
+
+	select {
+	case <-setOverrideReturned:
+	case <-time.After(30 * time.Second):
+		t.Error("could not set control request override within 30 seconds")
+		t.FailNow()
+	}
+
+	// Wait for the override to expire
+	time.Sleep(2 * overrideDuration)
+
+	// See whether override expired
+	fc.overrideMutex.RLock()
+	require.Equal(t, 0, len(fc.overrides), "override not removed")
+	fc.overrideMutex.RUnlock()
+
+	// Make sure that FlagsChanged was not held open by the observersMutex
+	require.True(t, d.flagsChangedCalled.Load(), "FlagsChanged not called")
+	require.True(t, d.flagsChangedReturned.Load(), "FlagsChanged did not return")
+
+	// Make sure that there's an observer registered for d.observerKey
+	fc.observersMutex.RLock()
+	observerForKeyFound := false
+	for _, keys := range fc.observers {
+		if slices.Contains(keys, newObserverKey) {
+			observerForKeyFound = true
+		}
+	}
+	fc.observersMutex.RUnlock()
+	require.True(t, observerForKeyFound, "new observer not successfully registered")
+}