From a6fb94857608ee6b8a39cfcefc7ef115bc77c416 Mon Sep 17 00:00:00 2001 From: Joseph Schorr Date: Thu, 22 Oct 2026 12:52:59 -0400 Subject: [PATCH 1/2] channelsd: surface and recover from a listener transport that dies for good A Slack Socket Mode loop that exits on a fatal error (invalid_auth, token_revoked, ...) emits no event, and the listener discarded the return value, so the bot went deaf while its Channel kept reporting Connected=True until channelsd was restarted. Listeners can now report a lost transport via Deps.TransportLost. channelsd marks the Channel Connected=False (reason TransportLost), which the monitoring watchers already publish to the monitoring Channel, and restarts the listener with per-Channel exponential backoff (5s doubling to 5m). Failed listener starts share the same backoff instead of retrying on every 5s tick. --- internal/cmd/channelsd/listeners.go | 107 ++++++++- .../listeners_transport_lost_test.go | 218 ++++++++++++++++++ pkg/apis/v1alpha1/conditions.go | 1 + pkg/channels/channelkinds/kind.go | 10 + pkg/channels/channelkinds/slack/listener.go | 33 ++- .../slack/listener_transport_lost_test.go | 91 ++++++++ 6 files changed, 451 insertions(+), 9 deletions(-) create mode 100644 internal/cmd/channelsd/listeners_transport_lost_test.go create mode 100644 pkg/channels/channelkinds/slack/listener_transport_lost_test.go diff --git a/internal/cmd/channelsd/listeners.go b/internal/cmd/channelsd/listeners.go index 4cc6b5ee..776f3ff1 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,27 @@ 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 } +// 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). +const ( + listenerRestartInitial = 5 * time.Second + listenerRestartMax = 5 * 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 +159,7 @@ func newChannelManager( cancels: map[string]context.CancelFunc{}, startedWiring: map[string]wiringSet{}, + restarts: flowcontrol.NewBackOffWithJitter(listenerRestartInitial, listenerRestartMax, 0.2), } } @@ -363,14 +385,25 @@ func (m *channelManager) reconcile(ctx context.Context, logger logr.Logger) { } 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") + if m.inRestartBackoff(key) { + // 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) + } sec, k, err := resolve.ForChannel(ctx, m.cli, ch) if err != nil { @@ -395,11 +428,15 @@ 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) + deps.TransportLost = func(err error) { + m.transportLost(ctx, listenerCtx, cancel, key, 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) + logger.Error(err, "start listener", "channel", key, "retryIn", retry.String()) m.patchConnected(ctx, ch, false, spiceboxv1alpha1.ReasonChannelListenerStartFailed, err.Error()) return } @@ -412,11 +449,70 @@ 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, key string, ch *spiceboxv1alpha1.Channel, err error) { + if listenerCtx.Err() != nil { + return // already stopped or shutting down; not a loss + } + retry := m.noteListenerFailure(key) + 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() +} + +// 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 string) time.Duration { + if m.restarts == nil { + return 0 + } + m.restarts.Next(key, m.restarts.Clock.Now()) + return m.restarts.Get(key) +} + +// inRestartBackoff reports whether key failed recently enough that a restart +// must wait. +func (m *channelManager) inRestartBackoff(key string) bool { + return m.restarts != nil && m.restarts.IsInBackOffSinceUpdate(key, m.restarts.Clock.Now()) +} + +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 +520,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..6a49278b --- /dev/null +++ b/internal/cmd/channelsd/listeners_transport_lost_test.go @@ -0,0 +1,218 @@ +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" + + 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") +} 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) + }) + } +} From 1fd963d363356ab0c7f73b6355a0dc8b99fd7f1a Mon Sep 17 00:00:00 2001 From: Joseph Schorr Date: Thu, 22 Oct 2026 15:23:06 -0400 Subject: [PATCH 2/2] channelsd: clear a Channel's listener restart backoff when it no longer applies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing ever cleared a key's restart backoff, so a Channel whose token was fixed, or that was deleted and recreated under the same name, could sit detached for up to 5m after the fix. The backoff now resets when the listener's inputs change (Channel UID, generation, or credentials Secret resourceVersion), when the Channel detaches (deleted, or invalid and not viable — including Channels whose Start never succeeded and so were never registered), and when a transport is lost after at least a minute of healthy uptime. --- internal/cmd/channelsd/listeners.go | 132 +++++++++++++++--- .../listeners_transport_lost_test.go | 85 +++++++++++ 2 files changed, 195 insertions(+), 22 deletions(-) diff --git a/internal/cmd/channelsd/listeners.go b/internal/cmd/channelsd/listeners.go index 776f3ff1..1a57d7cd 100644 --- a/internal/cmd/channelsd/listeners.go +++ b/internal/cmd/channelsd/listeners.go @@ -90,14 +90,29 @@ type channelManager struct { // 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). +// 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 @@ -342,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 @@ -370,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 { @@ -378,9 +398,15 @@ 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) } } @@ -394,17 +420,9 @@ func (m *channelManager) startListener(ctx context.Context, key string, ch *spic if exists && !isClosed(done) { return // already running } - if m.inRestartBackoff(key) { - // 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) - } + // 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) @@ -418,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 @@ -429,13 +458,14 @@ func (m *channelManager) startListener(ctx context.Context, key string, ch *spic m.mu.Unlock() listenerCtx, cancel := context.WithCancel(ctx) + attempt := listenerAttempt{key: key, inputs: inputs, at: m.now()} deps.TransportLost = func(err error) { - m.transportLost(ctx, listenerCtx, cancel, key, ch, err) + m.transportLost(ctx, listenerCtx, cancel, attempt, ch, err) } listener := k.NewListener(deps) if err := listener.Start(listenerCtx); err != nil { cancel() - retry := m.noteListenerFailure(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 @@ -473,11 +503,15 @@ func (m *channelManager) startListener(ctx context.Context, key string, ch *spic // 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, key string, ch *spiceboxv1alpha1.Channel, err error) { +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 } - retry := m.noteListenerFailure(key) + 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, @@ -485,20 +519,74 @@ func (m *channelManager) transportLost(ctx, listenerCtx context.Context, cancel 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 string) time.Duration { +func (m *channelManager) noteListenerFailure(key, inputs string) time.Duration { if m.restarts == nil { return 0 } - m.restarts.Next(key, m.restarts.Clock.Now()) + 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 that a restart -// must wait. -func (m *channelManager) inRestartBackoff(key string) bool { - return m.restarts != nil && m.restarts.IsInBackOffSinceUpdate(key, m.restarts.Clock.Now()) +// 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 { diff --git a/internal/cmd/channelsd/listeners_transport_lost_test.go b/internal/cmd/channelsd/listeners_transport_lost_test.go index 6a49278b..1b4f6bee 100644 --- a/internal/cmd/channelsd/listeners_transport_lost_test.go +++ b/internal/cmd/channelsd/listeners_transport_lost_test.go @@ -14,6 +14,7 @@ import ( 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" @@ -216,3 +217,87 @@ func TestTransportLostAfterStopIsIgnored(t *testing.T) { 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) + }) + } +}