diff --git a/internal/cmd/channelsd/listeners.go b/internal/cmd/channelsd/listeners.go index 4cc6b5ee..1a57d7cd 100644 --- a/internal/cmd/channelsd/listeners.go +++ b/internal/cmd/channelsd/listeners.go @@ -2,6 +2,7 @@ package main import ( "context" + "fmt" "strings" "sync" "time" @@ -9,6 +10,7 @@ import ( "github.com/go-logr/logr" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/flowcontrol" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/log" @@ -77,8 +79,42 @@ type channelManager struct { // of the process. reconcile compares this against the current wiring and // restarts only when something became available. startedWiring map[string]wiringSet + // dones holds each running listener's context Done channel. A listener + // whose transport died for good cancels its own context through + // Deps.TransportLost (see transportLost); startListener treats a closed + // one as "not running" and replaces it on the next reconcile tick. + dones map[string]<-chan struct{} + // restarts paces re-attempts for a Channel whose listener failed — a + // Start error, or a transport lost for good. A Slack app whose token + // answers invalid_auth would otherwise be re-dialled (flapping Connected, + // and paging the monitoring Channel each time) on every 5s reconcile tick. + // Nil disables pacing (struct-literal test managers). + restarts *flowcontrol.Backoff + // failedInputs records, per key, the listenerInputs the last failure ran + // with. A retry with different inputs — a rotated token, a spec edit, a + // Channel deleted and recreated under the same name — is a fresh attempt, + // not another round of the same failure, so it skips the backoff. Without + // this, the operator who fixed a revoked token would watch the Channel sit + // detached for up to listenerRestartMax and conclude the fix did not work. + // Its keys are also how reconcile finds failing Channels that have since + // vanished, since a Channel whose Start keeps failing is never in + // m.listeners. + failedInputs map[string]string } +// Listener restart pacing: the first retry comes after ~listenerRestartInitial, +// doubling up to listenerRestartMax. A Channel that goes 2*listenerRestartMax +// without a failure starts over from the initial delay (flowcontrol's expiry), +// and so does one whose transport stayed up for listenerHealthyAfter before it +// was lost: that is a new outage, not the same one flapping. Resetting at Start +// instead would defeat the backoff for a token that authenticates and is then +// rejected moments later. +const ( + listenerRestartInitial = 5 * time.Second + listenerRestartMax = 5 * time.Minute + listenerHealthyAfter = time.Minute +) + // wiringSet is a bitmask of the optional, late-wired collaborators a listener // can receive through Deps. Each is nil at construction time and set once, // milliseconds later, by internal/cmd/channelsd/run() — so the mask only ever grows. @@ -138,6 +174,7 @@ func newChannelManager( cancels: map[string]context.CancelFunc{}, startedWiring: map[string]wiringSet{}, + restarts: flowcontrol.NewBackOffWithJitter(listenerRestartInitial, listenerRestartMax, 0.2), } } @@ -320,6 +357,9 @@ func (m *channelManager) reconcile(ctx context.Context, logger logr.Logger) { m.patchConnected(ctx, ch, false, spiceboxv1alpha1.ReasonChannelSocketDetached, "Channel no longer Valid") } m.stopListener(logger, key) + // Whatever failed before the detach says nothing about the Channel + // that comes back Valid; it gets a prompt first attempt. + m.clearRestartBackoff(key) continue } // An invalid-but-viable Channel falls through to startListener on @@ -348,7 +388,9 @@ func (m *channelManager) reconcile(ctx context.Context, logger logr.Logger) { m.reconcileBento(ctx, key, ch) m.refreshListenerScopes(ctx, key, ch) } - // Stop listeners for Channels that no longer exist. + // Stop listeners for Channels that no longer exist, and drop the restart + // backoff of any that were failing — including ones whose Start never + // succeeded, which are in failedInputs but not m.listeners. m.mu.Lock() var gone []string for k := range m.listeners { @@ -356,22 +398,31 @@ func (m *channelManager) reconcile(ctx context.Context, logger logr.Logger) { gone = append(gone, k) } } + for k := range m.failedInputs { + if !seen[k] && m.listeners[k] == nil { + gone = append(gone, k) + } + } m.mu.Unlock() for _, k := range gone { m.stopListener(logger, k) + m.clearRestartBackoff(k) } } func (m *channelManager) startListener(ctx context.Context, key string, ch *spiceboxv1alpha1.Channel) { + logger := log.FromContext(ctx).WithName("channelmanager") + m.mu.Lock() _, exists := m.listeners[key] + done := m.dones[key] m.mu.Unlock() - if exists { + if exists && !isClosed(done) { return // already running } - logger := log.FromContext(ctx).WithName("channelmanager") - + // Resolved before the backoff check, not after: whether the Secret changed + // since the last failure is what decides if the backoff still applies. sec, k, err := resolve.ForChannel(ctx, m.cli, ch) if err != nil { logger.Error(err, "resolve channel", "channel", key) @@ -385,6 +436,17 @@ func (m *channelManager) startListener(ctx context.Context, key string, ch *spic logger.Info("listener skipped: client-hosted kind", "channel", key, "kind", k.Name()) return } + inputs := listenerInputs(ch, sec) + if m.inRestartBackoff(key, inputs) { + // A dead listener stays registered until its backoff elapses. That is + // harmless — its context is cancelled — and keeps "dead, waiting to + // retry" distinct from "never started". + return + } + if exists { + logger.Info("restarting listener whose transport was lost", "channel", key) + m.stopListener(logger, key) + } // Build the Deps and record which optional collaborators they carried in // ONE critical section, so the recorded set is exactly what the listener // got. Reading them separately would let a Set* land between the two and @@ -395,11 +457,16 @@ func (m *channelManager) startListener(ctx context.Context, key string, ch *spic wiring := m.optionalWiringLocked() m.mu.Unlock() - listener := k.NewListener(deps) listenerCtx, cancel := context.WithCancel(ctx) + attempt := listenerAttempt{key: key, inputs: inputs, at: m.now()} + deps.TransportLost = func(err error) { + m.transportLost(ctx, listenerCtx, cancel, attempt, ch, err) + } + listener := k.NewListener(deps) if err := listener.Start(listenerCtx); err != nil { cancel() - logger.Error(err, "start listener", "channel", key) + retry := m.noteListenerFailure(key, inputs) + logger.Error(err, "start listener", "channel", key, "retryIn", retry.String()) m.patchConnected(ctx, ch, false, spiceboxv1alpha1.ReasonChannelListenerStartFailed, err.Error()) return } @@ -412,11 +479,128 @@ func (m *channelManager) startListener(ctx context.Context, key string, ch *spic m.startedWiring = map[string]wiringSet{} } m.startedWiring[key] = wiring + if m.dones == nil { + m.dones = map[string]<-chan struct{}{} + } + m.dones[key] = listenerCtx.Done() m.mu.Unlock() + if listenerCtx.Err() != nil { + // The transport died between Start and here. transportLost has already + // stamped Connected=False; don't overwrite it with True. The next tick + // replaces the listener. + return + } logger.Info("listener started", "channel", key) m.patchConnected(ctx, ch, true, spiceboxv1alpha1.ReasonChannelSocketAttached, "") } +// transportLost is a listener's Deps.TransportLost: its transport died and +// will not recover on its own. It stamps Connected=False — which the +// operator's monitoring watchers turn into an Error event on the monitoring +// Channel — and only THEN cancels the listener's context, so the restart a +// later reconcile tick performs (and its Connected=True) cannot be overtaken by +// this False. It deliberately never touches m.listeners: Start may fire this +// before startListener has registered the listener, and the closed Done +// channel is what tells a later tick to replace it, once the restart backoff +// allows. +func (m *channelManager) transportLost(ctx, listenerCtx context.Context, cancel context.CancelFunc, attempt listenerAttempt, ch *spiceboxv1alpha1.Channel, err error) { + if listenerCtx.Err() != nil { + return // already stopped or shutting down; not a loss + } + key := attempt.key + if m.restarts != nil && m.now().Sub(attempt.at) >= listenerHealthyAfter { + m.restarts.Reset(key) // it was healthy; this is a new outage + } + retry := m.noteListenerFailure(key, attempt.inputs) + log.FromContext(ctx).WithName("channelmanager").Info("listener transport lost; restarting after backoff", + "channel", key, "err", err.Error(), "retryIn", retry.String()) + m.patchConnected(ctx, ch, false, spiceboxv1alpha1.ReasonChannelTransportLost, + fmt.Sprintf("connection lost and did not recover: %v. channelsd will restart the listener in about %s.", err, retry.Round(time.Second))) + cancel() +} + +// listenerAttempt is what transportLost needs to know about the start it is +// reporting on. +type listenerAttempt struct { + key string + inputs string // listenerInputs at start + at time.Time // when the start was attempted +} + +// listenerInputs fingerprints what a listener is built from: the Channel's +// identity and spec, and its credentials Secret's content. Any of them +// changing means a previous failure no longer predicts the next attempt. +func listenerInputs(ch *spiceboxv1alpha1.Channel, sec *corev1.Secret) string { + secretRV := "" + if sec != nil { + secretRV = sec.ResourceVersion + } + return fmt.Sprintf("%s/%d/%s", ch.UID, ch.Generation, secretRV) +} + +func (m *channelManager) now() time.Time { + if m.restarts == nil { + return time.Now() + } + return m.restarts.Clock.Now() +} + +// noteListenerFailure records a failed start or a lost transport for key and +// returns how long until the next attempt is allowed. +func (m *channelManager) noteListenerFailure(key, inputs string) time.Duration { + if m.restarts == nil { + return 0 + } + m.mu.Lock() + if m.failedInputs == nil { + m.failedInputs = map[string]string{} + } + m.failedInputs[key] = inputs + m.mu.Unlock() + m.restarts.Next(key, m.now()) + return m.restarts.Get(key) +} + +// inRestartBackoff reports whether key failed recently enough, with these same +// inputs, that a restart must wait. A change of inputs clears the backoff. +func (m *channelManager) inRestartBackoff(key, inputs string) bool { + if m.restarts == nil { + return false + } + m.mu.Lock() + failedWith, failed := m.failedInputs[key] + m.mu.Unlock() + if failed && failedWith != inputs { + m.clearRestartBackoff(key) + return false + } + return m.restarts.IsInBackOffSinceUpdate(key, m.now()) +} + +// clearRestartBackoff forgets key's failure history, so its next attempt runs +// immediately and a later failure starts again from listenerRestartInitial. +func (m *channelManager) clearRestartBackoff(key string) { + if m.restarts == nil { + return + } + m.mu.Lock() + delete(m.failedInputs, key) + m.mu.Unlock() + m.restarts.Reset(key) +} + +func isClosed(ch <-chan struct{}) bool { + if ch == nil { + return false + } + select { + case <-ch: + return true + default: + return false + } +} + func (m *channelManager) stopListener(logger logr.Logger, key string) { m.mu.Lock() l, ok := m.listeners[key] @@ -424,6 +608,7 @@ func (m *channelManager) stopListener(logger logr.Logger, key string) { delete(m.listeners, key) delete(m.cancels, key) delete(m.startedWiring, key) + delete(m.dones, key) m.mu.Unlock() if !ok { return diff --git a/internal/cmd/channelsd/listeners_transport_lost_test.go b/internal/cmd/channelsd/listeners_transport_lost_test.go new file mode 100644 index 00000000..1b4f6bee --- /dev/null +++ b/internal/cmd/channelsd/listeners_transport_lost_test.go @@ -0,0 +1,303 @@ +package main + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/flowcontrol" + clocktesting "k8s.io/utils/clock/testing" + "sigs.k8s.io/controller-runtime/pkg/client" + + spiceboxv1alpha1 "github.com/authzed/openagentprimitives/pkg/apis/v1alpha1" + "github.com/authzed/openagentprimitives/pkg/channels/channelkinds" + "github.com/authzed/openagentprimitives/pkg/channels/channelkinds/registry" +) + +const lostKindName = "transportlosttest" + +var registerLostKindOnce sync.Once + +// lostKind is the fake kind with a listener that hands its Deps back to the +// test, so the test can fire Deps.TransportLost the way a real transport would. +type lostKind struct{ channelkinds.Kind } + +func (*lostKind) Name() string { return lostKindName } + +func (*lostKind) NewListener(deps channelkinds.Deps) channelkinds.Listener { + l := &lostListener{deps: deps} + lostListeners.Lock() + lostListeners.all = append(lostListeners.all, l) + lostListeners.Unlock() + return l +} + +// lostListenerHook, when set, runs inside Start — before startListener has +// registered the listener — to exercise a transport that dies immediately. +// Its return value is Start's. +var lostListenerHook func(deps channelkinds.Deps) error + +var lostListeners struct { + sync.Mutex + all []*lostListener +} + +type lostListener struct{ deps channelkinds.Deps } + +func (l *lostListener) Start(context.Context) error { + if lostListenerHook != nil { + return lostListenerHook(l.deps) + } + return nil +} + +func (*lostListener) Stop(context.Context) error { return nil } + +func registerLostKind(t *testing.T) { + t.Helper() + base, ok := registry.Get("fake") + require.True(t, ok, "the fake kind must be registered to embed it") + registerLostKindOnce.Do(func() { registry.Register(&lostKind{Kind: base}) }) + lostListeners.Lock() + lostListeners.all = nil + lostListeners.Unlock() + t.Cleanup(func() { lostListenerHook = nil }) +} + +func lostListenerCount() int { + lostListeners.Lock() + defer lostListeners.Unlock() + return len(lostListeners.all) +} + +func lastLostListener(t *testing.T) *lostListener { + t.Helper() + lostListeners.Lock() + defer lostListeners.Unlock() + require.NotEmpty(t, lostListeners.all, "no listener was constructed") + return lostListeners.all[len(lostListeners.all)-1] +} + +func newLostManager(t *testing.T) (*channelManager, *spiceboxv1alpha1.Channel, func() *metav1.Condition) { + t.Helper() + registerLostKind(t) + ch := &spiceboxv1alpha1.Channel{ + ObjectMeta: metav1.ObjectMeta{Namespace: "demo-ns", Name: "demo-channel"}, + Spec: spiceboxv1alpha1.ChannelSpec{ + Kind: lostKindName, + AgentClass: "demo-agent", + CredentialsRef: spiceboxv1alpha1.ChannelCredentialsRef{SecretName: "demo-creds"}, + }, + } + sec := &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: "demo-ns", Name: "demo-creds"}} + cli := newClobberK8s(t, ch, sec) + m := &channelManager{cli: cli, listeners: map[string]channelkinds.Listener{}, cancels: map[string]context.CancelFunc{}} + t.Cleanup(func() { m.Stop(logr.Discard()) }) + connected := func() *metav1.Condition { + return channelCondition(t, cli, "demo-ns", "demo-channel", spiceboxv1alpha1.ChannelConditionConnected) + } + return m, ch, connected +} + +const lostKey = "demo-ns/demo-channel" + +func TestTransportLostMarksDisconnectedAndNextTickRestarts(t *testing.T) { + m, ch, connected := newLostManager(t) + ctx := context.Background() + + m.startListener(ctx, lostKey, ch) + first := lastLostListener(t) + require.Equal(t, metav1.ConditionTrue, connected().Status, "precondition: listener attached") + + first.deps.TransportLost(errors.New("slack socket mode: invalid_auth")) + + got := connected() + assert.Equal(t, metav1.ConditionFalse, got.Status) + assert.Equal(t, spiceboxv1alpha1.ReasonChannelTransportLost, got.Reason) + assert.Contains(t, got.Message, "invalid_auth", "the cause must reach the condition (and so the monitoring event)") + + // The next reconcile tick's startListener must replace the dead listener, + // not early-return because the key is still in the map. + m.startListener(ctx, lostKey, ch) + assert.NotSame(t, first, lastLostListener(t), "the dead listener must be replaced") + assert.Equal(t, metav1.ConditionTrue, connected().Status, "a successful restart re-attaches") + + // A further tick with a healthy listener is a no-op. + second := lastLostListener(t) + m.startListener(ctx, lostKey, ch) + assert.Same(t, second, lastLostListener(t), "a healthy listener must not be restarted") +} + +func TestTransportLostDuringStartIsNotOverwrittenByConnectedTrue(t *testing.T) { + m, ch, connected := newLostManager(t) + ctx := context.Background() + lostListenerHook = func(deps channelkinds.Deps) error { + deps.TransportLost(errors.New("token_revoked")) + return nil + } + + m.startListener(ctx, lostKey, ch) + + got := connected() + require.NotNil(t, got) + assert.Equal(t, metav1.ConditionFalse, got.Status, "startListener must not stamp True over a loss reported during Start") + assert.Equal(t, spiceboxv1alpha1.ReasonChannelTransportLost, got.Reason) + + lostListenerHook = nil + first := lastLostListener(t) + m.startListener(ctx, lostKey, ch) + assert.NotSame(t, first, lastLostListener(t), "the next tick must still replace it") + assert.Equal(t, metav1.ConditionTrue, connected().Status) +} + +// TestFailedListenersRestartWithExponentialBackoff pins that a fatal failure +// is retried — never given up on — but paced: 5s, then 10s, for both a lost +// transport and a Start that errors. +func TestFailedListenersRestartWithExponentialBackoff(t *testing.T) { + m, ch, connected := newLostManager(t) + clk := clocktesting.NewFakeClock(time.Now()) + m.restarts = flowcontrol.NewFakeBackOff(listenerRestartInitial, listenerRestartMax, clk) + ctx := context.Background() + + m.startListener(ctx, lostKey, ch) + lastLostListener(t).deps.TransportLost(errors.New("invalid_auth")) + assert.Contains(t, connected().Message, "in about 5s", "the condition says when the retry comes") + + built := lostListenerCount() + m.startListener(ctx, lostKey, ch) + assert.Equal(t, built, lostListenerCount(), "no restart inside the first backoff window") + + clk.Step(listenerRestartInitial) + m.startListener(ctx, lostKey, ch) + require.Equal(t, built+1, lostListenerCount(), "restart once the first window elapses") + assert.Equal(t, metav1.ConditionTrue, connected().Status) + + // Second consecutive failure — this time Start itself errors on the retry. + lastLostListener(t).deps.TransportLost(errors.New("invalid_auth")) + clk.Step(listenerRestartInitial) + built = lostListenerCount() + m.startListener(ctx, lostKey, ch) + assert.Equal(t, built, lostListenerCount(), "the window doubled to 10s: 5s is too early") + + clk.Step(listenerRestartInitial) + lostListenerHook = func(channelkinds.Deps) error { return errors.New("auth.test: invalid_auth") } + m.startListener(ctx, lostKey, ch) + require.Equal(t, built+1, lostListenerCount(), "retry at 10s") + assert.Equal(t, spiceboxv1alpha1.ReasonChannelListenerStartFailed, connected().Reason) + + lostListenerHook = nil + built = lostListenerCount() + clk.Step(listenerRestartInitial * 2) + m.startListener(ctx, lostKey, ch) + assert.Equal(t, built, lostListenerCount(), "a failed Start doubles the window again, to 20s") + + clk.Step(listenerRestartInitial * 2) + m.startListener(ctx, lostKey, ch) + assert.Equal(t, built+1, lostListenerCount(), "retry at 20s") + assert.Equal(t, metav1.ConditionTrue, connected().Status, "and it recovers") +} + +func TestTransportLostAfterStopIsIgnored(t *testing.T) { + m, ch, connected := newLostManager(t) + ctx := context.Background() + + m.startListener(ctx, lostKey, ch) + l := lastLostListener(t) + m.stopListener(logr.Discard(), lostKey) + + l.deps.TransportLost(errors.New("use of closed network connection")) + + assert.Equal(t, metav1.ConditionTrue, connected().Status, + "a transport that dies because channelsd stopped it is not a loss") +} + +func withFakeBackoff(m *channelManager) *clocktesting.FakeClock { + clk := clocktesting.NewFakeClock(time.Now()) + m.restarts = flowcontrol.NewFakeBackOff(listenerRestartInitial, listenerRestartMax, clk) + return clk +} + +// TestRestartBackoffSkippedWhenInputsChange pins that fixing what failed — here +// a rotated credentials Secret — gets an immediate retry rather than waiting +// out a backoff earned by the old inputs. +func TestRestartBackoffSkippedWhenInputsChange(t *testing.T) { + m, ch, connected := newLostManager(t) + withFakeBackoff(m) + ctx := context.Background() + lostListenerHook = func(channelkinds.Deps) error { return errors.New("auth.test: invalid_auth") } + + m.startListener(ctx, lostKey, ch) + built := lostListenerCount() + m.startListener(ctx, lostKey, ch) + require.Equal(t, built, lostListenerCount(), "precondition: same inputs wait out the backoff") + + var sec corev1.Secret + require.NoError(t, m.cli.Get(ctx, client.ObjectKey{Namespace: "demo-ns", Name: "demo-creds"}, &sec)) + sec.Data = map[string][]byte{"token": []byte("rotated")} + require.NoError(t, m.cli.Update(ctx, &sec)) + lostListenerHook = nil + + m.startListener(ctx, lostKey, ch) + assert.Equal(t, built+1, lostListenerCount(), "a changed Secret retries inside the old window") + assert.Equal(t, metav1.ConditionTrue, connected().Status) +} + +// TestRestartBackoffResetsAfterHealthyRun pins that a loss after a healthy run +// starts over at the initial delay instead of inheriting an earlier outage's. +func TestRestartBackoffResetsAfterHealthyRun(t *testing.T) { + m, ch, connected := newLostManager(t) + clk := withFakeBackoff(m) + ctx := context.Background() + + m.startListener(ctx, lostKey, ch) + lastLostListener(t).deps.TransportLost(errors.New("invalid_auth")) + clk.Step(listenerRestartInitial) + m.startListener(ctx, lostKey, ch) + lastLostListener(t).deps.TransportLost(errors.New("invalid_auth")) + require.Contains(t, connected().Message, "in about 10s", "precondition: a quick second loss doubles the window") + + clk.Step(2 * listenerRestartInitial) + m.startListener(ctx, lostKey, ch) + require.Equal(t, metav1.ConditionTrue, connected().Status) + clk.Step(listenerHealthyAfter) + lastLostListener(t).deps.TransportLost(errors.New("token_revoked")) + assert.Contains(t, connected().Message, "in about 5s", "a loss after a healthy run is a new outage") +} + +// TestRestartBackoffClearedWhenChannelDetaches pins that a Channel which stops +// being served — deleted, or detached as invalid — loses its failure history, +// so the Channel that comes back attaches on the next tick. Both cases use a +// Start that never succeeded, which is never in m.listeners. +func TestRestartBackoffClearedWhenChannelDetaches(t *testing.T) { + for name, detach := range map[string]func(context.Context, client.Client) error{ + "Channel deleted: backoff cleared": func(ctx context.Context, c client.Client) error { + return c.Delete(ctx, &spiceboxv1alpha1.Channel{ObjectMeta: metav1.ObjectMeta{Namespace: "demo-ns", Name: "demo-channel"}}) + }, + "credentials Secret deleted, Channel detached as invalid: backoff cleared": func(ctx context.Context, c client.Client) error { + return c.Delete(ctx, &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: "demo-ns", Name: "demo-creds"}}) + }, + } { + t.Run(name, func(t *testing.T) { + m, ch, _ := newLostManager(t) + clk := withFakeBackoff(m) + ctx := context.Background() + lostListenerHook = func(channelkinds.Deps) error { return errors.New("auth.test: invalid_auth") } + + m.startListener(ctx, lostKey, ch) + require.True(t, m.restarts.IsInBackOffSinceUpdate(lostKey, clk.Now()), "precondition: the failed Start is backing off") + + require.NoError(t, detach(ctx, m.cli)) + m.reconcile(ctx, logr.Discard()) + + assert.False(t, m.restarts.IsInBackOffSinceUpdate(lostKey, clk.Now())) + assert.Empty(t, m.failedInputs) + }) + } +} diff --git a/pkg/apis/v1alpha1/conditions.go b/pkg/apis/v1alpha1/conditions.go index 8f91ee88..2622ab88 100644 --- a/pkg/apis/v1alpha1/conditions.go +++ b/pkg/apis/v1alpha1/conditions.go @@ -742,6 +742,7 @@ const ( ReasonChannelSocketDetached = "SocketDetached" // set by channelsd on Connected=False ReasonChannelDependenciesUnavailable = "DependenciesUnavailable" // memory or SpiceDB unreachable ReasonChannelListenerStartFailed = "ListenerStartFailed" // set by channelsd when listener.Start fails + ReasonChannelTransportLost = "TransportLost" // set by channelsd when a running listener's transport died without recovering ReasonChannelScopesGranted = "ScopesGranted" // set by channelsd on ScopesValid=True ReasonChannelMissingScopes = "MissingScopes" // set by channelsd on ScopesValid=False; message lists the missing scopes ReasonChannelDeliverySucceeded = "DeliverySucceeded" // set by channelsd's relay on Deliverable=True diff --git a/pkg/channels/channelkinds/kind.go b/pkg/channels/channelkinds/kind.go index ac5bb817..424a9af6 100644 --- a/pkg/channels/channelkinds/kind.go +++ b/pkg/channels/channelkinds/kind.go @@ -589,6 +589,16 @@ type Deps struct { // 0 ⇒ DefaultApproverFanoutLimit. Set from channelsd's // --approver-fanout-limit. ApproverFanoutLimit int + + // TransportLost is called by a running Listener whose transport has died + // and will NOT recover on its own — e.g. a Slack Socket Mode loop that + // exited on a fatal error instead of reconnecting. Without it such a + // listener is a silent zombie: still registered, Connected still True, no + // inbound ever again. channelsd stamps Connected=False (which the + // monitoring watchers surface) and replaces the listener on its next + // reconcile tick. Must not be called for a Stop- or ctx-initiated + // shutdown. May be called before Start returns. Optional; nil-safe. + TransportLost func(err error) } // DefaultApproverFanoutLimit is the default for Deps.ApproverFanoutLimit: diff --git a/pkg/channels/channelkinds/slack/listener.go b/pkg/channels/channelkinds/slack/listener.go index 507a320e..2015ab98 100644 --- a/pkg/channels/channelkinds/slack/listener.go +++ b/pkg/channels/channelkinds/slack/listener.go @@ -594,10 +594,35 @@ func (l *slackListener) run(ctx context.Context) { } }() - // RunContext blocks and handles reconnection internally. - // We ignore the error — on ctx.Done() it returns a context error, which - // is expected. Genuine disconnects are reconnected by socketmode internally. - _ = l.src.RunContext(ctx) + // RunContext blocks and handles transient disconnects internally. It + // returns for good only on shutdown (ctx cancelled) or on an error it + // deems fatal — invalid_auth, token_revoked, account_inactive, not_authed, + // or a 404 from apps.connections.open — and emits NO socketmode event for + // most of those. Slack can answer that way transiently while degraded, so + // the return value is the only signal: discarding it once left a listener + // registered, Connected=True, and deaf until channelsd was restarted. + err := l.src.RunContext(ctx) + if ctx.Err() != nil || l.stopped() { + return + } + if err == nil { + err = fmt.Errorf("socket mode loop exited without an error") + } + log.FromContext(ctx).Info("slack listener: socket mode connection lost and not reconnecting", + "channel", l.deps.Channel.Namespace+"/"+l.deps.Channel.Name, "err", err.Error()) + if l.deps.TransportLost != nil { + l.deps.TransportLost(fmt.Errorf("slack socket mode: %w", err)) + } +} + +// stopped reports whether Stop has been called. +func (l *slackListener) stopped() bool { + select { + case <-l.stop: + return true + default: + return false + } } // handle dispatches a single socketmode event. diff --git a/pkg/channels/channelkinds/slack/listener_transport_lost_test.go b/pkg/channels/channelkinds/slack/listener_transport_lost_test.go new file mode 100644 index 00000000..3f45cca8 --- /dev/null +++ b/pkg/channels/channelkinds/slack/listener_transport_lost_test.go @@ -0,0 +1,91 @@ +package slack + +import ( + "context" + "errors" + "testing" + + "github.com/slack-go/slack/socketmode" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + spiceboxv1alpha1 "github.com/authzed/openagentprimitives/pkg/apis/v1alpha1" + "github.com/authzed/openagentprimitives/pkg/channels/channelkinds" +) + +// exitingSource is a socketSource whose RunContext does whatever runFn says, +// standing in for socketmode.Client.RunContext returning (it only ever does on +// shutdown or a fatal connect error). +type exitingSource struct { + runFn func(ctx context.Context) error +} + +func (exitingSource) Events() <-chan socketmode.Event { return nil } +func (s exitingSource) RunContext(ctx context.Context) error { return s.runFn(ctx) } +func (exitingSource) Ack(socketmode.Request, ...interface{}) {} + +func TestRunReportsTransportLostOnlyWhenNotShuttingDown(t *testing.T) { + cases := []struct { + name string + // runFn receives the listener and the cancel for its ctx so a case can + // simulate a shutdown racing the return. + runFn func(l *slackListener, cancel context.CancelFunc) func(context.Context) error + wantErr string // "" = TransportLost must NOT be called + }{ + { + name: "fatal connect error: TransportLost called with the cause", + runFn: func(*slackListener, context.CancelFunc) func(context.Context) error { + return func(context.Context) error { return errors.New("invalid_auth") } + }, + wantErr: "slack socket mode: invalid_auth", + }, + { + name: "nil return without shutdown: TransportLost still called", + runFn: func(*slackListener, context.CancelFunc) func(context.Context) error { + return func(context.Context) error { return nil } + }, + wantErr: "exited without an error", + }, + { + name: "ctx cancelled: shutdown, TransportLost not called", + runFn: func(_ *slackListener, cancel context.CancelFunc) func(context.Context) error { + return func(ctx context.Context) error { cancel(); return ctx.Err() } + }, + }, + { + name: "Stop called: shutdown, TransportLost not called", + runFn: func(l *slackListener, _ context.CancelFunc) func(context.Context) error { + return func(ctx context.Context) error { + require.NoError(t, l.Stop(ctx)) + return errors.New("use of closed network connection") + } + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + var lost []error + l := &slackListener{ + deps: channelkinds.Deps{ + Channel: &spiceboxv1alpha1.Channel{ObjectMeta: metav1.ObjectMeta{Namespace: "demo-ns", Name: "demo-channel"}}, + TransportLost: func(err error) { lost = append(lost, err) }, + }, + stop: make(chan struct{}), + } + l.src = exitingSource{runFn: tc.runFn(l, cancel)} + + l.run(ctx) // returns once RunContext does + + if tc.wantErr == "" { + assert.Empty(t, lost, "a shutdown must not be reported as a lost transport") + return + } + require.Len(t, lost, 1, "a lost transport must be reported exactly once") + assert.Contains(t, lost[0].Error(), tc.wantErr) + }) + } +}