mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 20:17:20 +00:00
fix(channels): bound Reload Start with timeout; preserve caller ctx for long-lived goroutines
A single hung ch.Start() (e.g. Telegram GetMe stalled on DNS/TCP) blocks InstanceLoader.mu and wedges every subsequent Reload(), leaving new channels stuck in "checking" forever. Wrap Start() in a goroutine bounded by reloadStartTimeout (90s) so Reload can move on to the next instance. On timeout, Stop the channel in a bounded window, record a typed failure for the UI, and asynchronously drain the late-returning Start to detect channels that ignore context cancellation. Critically, pass the caller's ctx — not a timeout-wrapped ctx — into ch.Start. Channels routinely derive long-running goroutines from the ctx they receive (Telegram's pollCtx is WithCancel(ctx)). A timeout- wrapped ctx would silently kill polling 90s after a successful Reload. Add regression tests covering both the hang-then-timeout path and the caller-context preservation invariant. Closes #914
This commit is contained in:
1 parent
3fe5ae0b50
commit
ce620c87ef
2 files changed
+265
-7
No files matched your search
@@ -18,6 +18,13 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
// reloadStartTimeout bounds how long Reload() will wait for a single channel's
|
||||
// Start() to return before abandoning it and continuing with the rest.
|
||||
// Sits above Telegram's probeOverallTimeout (60s) so well-behaved channels
|
||||
// always get a chance to finish before being given up on.
|
||||
// var (not const) so tests can shrink it without waiting a real minute+.
|
||||
var reloadStartTimeout = 90 * time.Second
|
||||
|
||||
// ChannelFactory creates a Channel from DB instance data.
|
||||
// name: channel name (registered in Manager, used in session keys).
|
||||
// creds: decrypted credentials JSON (token, API keys, etc.).
|
||||
@@ -325,17 +332,72 @@ func (l *InstanceLoader) loadInstance(ctx context.Context, inst store.ChannelIns
|
||||
l.manager.RegisterChannel(inst.Name, ch)
|
||||
|
||||
// Start the channel if requested (Reload path). LoadAll defers to StartAll.
|
||||
// Bound the wait so one hung Start() can't block Reload()'s mutex and wedge
|
||||
// every subsequent reload. Important: we pass the caller's ctx (not a
|
||||
// timeout-wrapped one) to ch.Start so long-running goroutines the channel
|
||||
// derives from it — e.g. Telegram's pollCtx — are not cancelled out from
|
||||
// under a successful start.
|
||||
if autoStart {
|
||||
if err := ch.Start(ctx); err != nil {
|
||||
l.manager.recordChannelStartFailure(inst.Name, ch, "", err)
|
||||
slog.Error("channel instance start failed", "name", inst.Name, "error", err)
|
||||
// Still registered — will show as not running.
|
||||
} else {
|
||||
l.manager.RecordHealth(inst.Name, snapshotChannelHealth(ch))
|
||||
}
|
||||
l.startChannelWithTimeout(ctx, inst, ch)
|
||||
}
|
||||
|
||||
slog.Info("channel instance loaded",
|
||||
"name", inst.Name, "type", inst.ChannelType, "agent_id", inst.AgentID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// startChannelWithTimeout runs ch.Start(ctx) in a goroutine and waits up to
|
||||
// reloadStartTimeout for it to return. On timeout we stop the partially-started
|
||||
// channel and record a failure so Reload() can move on to the next instance.
|
||||
//
|
||||
// ctx is passed through unchanged: channels routinely derive long-lived
|
||||
// goroutines (e.g. Telegram long-polling) from this context and must keep
|
||||
// running after Start returns. A late-returning Start — i.e. one that ignores
|
||||
// the caller ctx entirely — is drained asynchronously so its goroutine doesn't
|
||||
// block forever on the send to startErr. If it eventually reports success,
|
||||
// we've already called Stop, which is idempotent across channel impls.
|
||||
func (l *InstanceLoader) startChannelWithTimeout(ctx context.Context, inst store.ChannelInstanceData, ch Channel) {
|
||||
startErr := make(chan error, 1)
|
||||
go func() { startErr <- ch.Start(ctx) }()
|
||||
|
||||
timer := time.NewTimer(reloadStartTimeout)
|
||||
defer timer.Stop()
|
||||
|
||||
select {
|
||||
case err := <-startErr:
|
||||
if err != nil {
|
||||
l.manager.recordChannelStartFailure(inst.Name, ch, "", err)
|
||||
slog.Error("channel instance start failed",
|
||||
"name", inst.Name, "type", inst.ChannelType, "error", err)
|
||||
return
|
||||
}
|
||||
l.manager.RecordHealth(inst.Name, snapshotChannelHealth(ch))
|
||||
|
||||
case <-timer.C:
|
||||
// Stop the channel in a bounded window so we don't trade one hang for another.
|
||||
stopCtx, stopCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
if err := ch.Stop(stopCtx); err != nil {
|
||||
slog.Warn("failed to stop timed-out channel",
|
||||
"name", inst.Name, "type", inst.ChannelType, "error", err)
|
||||
}
|
||||
stopCancel()
|
||||
|
||||
timeoutErr := fmt.Errorf("start timed out after %s (type=%s)", reloadStartTimeout, inst.ChannelType)
|
||||
l.manager.recordChannelStartFailure(inst.Name, ch, "", timeoutErr)
|
||||
slog.Error("channel instance start timed out",
|
||||
"name", inst.Name, "type", inst.ChannelType, "timeout", reloadStartTimeout)
|
||||
|
||||
// Drain the late-returning Start so its goroutine can exit.
|
||||
// Logged so operators can spot channels that ignore context cancellation.
|
||||
go func() {
|
||||
err := <-startErr
|
||||
if err != nil {
|
||||
slog.Warn("channel instance start returned after timeout",
|
||||
"name", inst.Name, "type", inst.ChannelType, "error", err)
|
||||
return
|
||||
}
|
||||
slog.Warn("channel instance start succeeded after timeout; already stopped",
|
||||
"name", inst.Name, "type", inst.ChannelType)
|
||||
}()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
package channels
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
// timeoutTestChannel is a minimal Channel implementation used by the Reload
|
||||
// timeout regression tests. It intentionally does NOT embed *BaseChannel so
|
||||
// optional interfaces (SetAgentID, SetTenantID, ...) aren't promoted — that
|
||||
// keeps loadInstance on the "no agent store needed" path.
|
||||
type timeoutTestChannel struct {
|
||||
base *BaseChannel
|
||||
release chan struct{}
|
||||
hang bool
|
||||
captureCtx func(context.Context)
|
||||
startReturn chan error
|
||||
stopCalls atomic.Int32
|
||||
}
|
||||
|
||||
func newTimeoutTestChannel(name, channelType string, hang bool) *timeoutTestChannel {
|
||||
b := NewBaseChannel(name, bus.New(), nil)
|
||||
b.SetType(channelType)
|
||||
return &timeoutTestChannel{
|
||||
base: b,
|
||||
release: make(chan struct{}),
|
||||
hang: hang,
|
||||
startReturn: make(chan error, 1),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *timeoutTestChannel) Name() string { return c.base.Name() }
|
||||
func (c *timeoutTestChannel) Type() string { return c.base.Type() }
|
||||
func (c *timeoutTestChannel) IsRunning() bool { return c.base.IsRunning() }
|
||||
func (c *timeoutTestChannel) IsAllowed(senderID string) bool { return c.base.IsAllowed(senderID) }
|
||||
func (c *timeoutTestChannel) Send(context.Context, bus.OutboundMessage) error { return nil }
|
||||
func (c *timeoutTestChannel) SetType(t string) { c.base.SetType(t) }
|
||||
func (c *timeoutTestChannel) HealthSnapshot() ChannelHealth { return c.base.HealthSnapshot() }
|
||||
|
||||
func (c *timeoutTestChannel) Start(ctx context.Context) error {
|
||||
if c.captureCtx != nil {
|
||||
c.captureCtx(ctx)
|
||||
}
|
||||
if !c.hang {
|
||||
c.base.MarkHealthy("Connected")
|
||||
return nil
|
||||
}
|
||||
// Hang until released or ctx done (covers both honor-ctx and ignore-ctx cases
|
||||
// — whichever fires first).
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
c.startReturn <- ctx.Err()
|
||||
return ctx.Err()
|
||||
case <-c.release:
|
||||
c.startReturn <- nil
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (c *timeoutTestChannel) Stop(context.Context) error {
|
||||
c.stopCalls.Add(1)
|
||||
c.base.MarkStopped("Stopped")
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkFailed delegates to the embedded base so recordChannelStartFailureLocked
|
||||
// can transition the channel from "stopped" → "failed" after the timeout cleanup
|
||||
// (real channels embed *BaseChannel and inherit this for free).
|
||||
func (c *timeoutTestChannel) MarkFailed(summary, detail string, kind ChannelFailureKind, retryable bool) {
|
||||
c.base.MarkFailed(summary, detail, kind, retryable)
|
||||
}
|
||||
|
||||
// withShortReloadTimeout shrinks the package-level timeout for the duration
|
||||
// of a test so we don't wait the real 90s.
|
||||
func withShortReloadTimeout(t *testing.T, d time.Duration) {
|
||||
t.Helper()
|
||||
orig := reloadStartTimeout
|
||||
reloadStartTimeout = d
|
||||
t.Cleanup(func() { reloadStartTimeout = orig })
|
||||
}
|
||||
|
||||
// TestLoadInstance_HungStartDoesNotBlock verifies that a Start() that never
|
||||
// returns is abandoned after reloadStartTimeout so Reload() can proceed to
|
||||
// other channels instead of deadlocking on the loader mutex.
|
||||
func TestLoadInstance_HungStartDoesNotBlock(t *testing.T) {
|
||||
withShortReloadTimeout(t, 200*time.Millisecond)
|
||||
|
||||
msgBus := bus.New()
|
||||
mgr := NewManager(msgBus)
|
||||
loader := NewInstanceLoader(nil, nil, mgr, msgBus, nil)
|
||||
|
||||
hung := newTimeoutTestChannel("telegram-hung", TypeTelegram, true)
|
||||
loader.RegisterFactory(TypeTelegram, func(string, json.RawMessage, json.RawMessage, *bus.MessageBus, store.PairingStore) (Channel, error) {
|
||||
return hung, nil
|
||||
})
|
||||
|
||||
done := make(chan error, 1)
|
||||
start := time.Now()
|
||||
loader.mu.Lock()
|
||||
go func() {
|
||||
defer loader.mu.Unlock()
|
||||
done <- loader.loadInstance(context.Background(), store.ChannelInstanceData{
|
||||
Name: "telegram-hung",
|
||||
ChannelType: TypeTelegram,
|
||||
}, true)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("loadInstance returned error: %v", err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("loadInstance did not return within 3s — timeout path is broken")
|
||||
}
|
||||
|
||||
if elapsed := time.Since(start); elapsed < 150*time.Millisecond {
|
||||
t.Fatalf("returned too fast (%s); timeout should have waited", elapsed)
|
||||
}
|
||||
|
||||
status := mgr.GetStatus()["telegram-hung"].(ChannelHealth)
|
||||
if status.State != ChannelHealthStateFailed {
|
||||
t.Fatalf("expected failed state, got %q", status.State)
|
||||
}
|
||||
if hung.stopCalls.Load() == 0 {
|
||||
t.Fatal("expected Stop() to be called on timed-out channel for cleanup")
|
||||
}
|
||||
|
||||
// Drain the start goroutine so it doesn't leak past the test.
|
||||
close(hung.release)
|
||||
select {
|
||||
case <-hung.startReturn:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("start goroutine did not drain after release")
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadInstance_FastStartPreservesCallerContext verifies the critical fix:
|
||||
// channels derive long-running goroutines from the ctx they receive in Start().
|
||||
// We must pass the caller's ctx through unchanged, not a timeout-wrapped ctx,
|
||||
// otherwise a successful Start() has its long-lived work cancelled when the
|
||||
// timeout fires or cancel runs — silently breaking polling after Reload.
|
||||
func TestLoadInstance_FastStartPreservesCallerContext(t *testing.T) {
|
||||
withShortReloadTimeout(t, 200*time.Millisecond)
|
||||
|
||||
msgBus := bus.New()
|
||||
mgr := NewManager(msgBus)
|
||||
loader := NewInstanceLoader(nil, nil, mgr, msgBus, nil)
|
||||
|
||||
var receivedCtx context.Context
|
||||
ch := newTimeoutTestChannel("telegram-fast", TypeTelegram, false)
|
||||
ch.captureCtx = func(ctx context.Context) { receivedCtx = ctx }
|
||||
loader.RegisterFactory(TypeTelegram, func(string, json.RawMessage, json.RawMessage, *bus.MessageBus, store.PairingStore) (Channel, error) {
|
||||
return ch, nil
|
||||
})
|
||||
|
||||
parent, cancelParent := context.WithCancel(context.Background())
|
||||
defer cancelParent()
|
||||
|
||||
loader.mu.Lock()
|
||||
err := loader.loadInstance(parent, store.ChannelInstanceData{
|
||||
Name: "telegram-fast",
|
||||
ChannelType: TypeTelegram,
|
||||
}, true)
|
||||
loader.mu.Unlock()
|
||||
if err != nil {
|
||||
t.Fatalf("loadInstance error: %v", err)
|
||||
}
|
||||
|
||||
if receivedCtx == nil {
|
||||
t.Fatal("Start did not receive a ctx")
|
||||
}
|
||||
// Wait past the timeout window. If Start had been handed a timeout ctx,
|
||||
// receivedCtx.Done() would be closed by now. It MUST still be open,
|
||||
// otherwise long-running polling goroutines (Telegram) would die.
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
select {
|
||||
case <-receivedCtx.Done():
|
||||
t.Fatal("ctx passed to Start was cancelled after timeout window — would kill derived polling goroutines")
|
||||
default:
|
||||
}
|
||||
|
||||
// Cancelling the parent ctx must still propagate (caller retains control).
|
||||
cancelParent()
|
||||
select {
|
||||
case <-receivedCtx.Done():
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("parent cancel did not propagate to Start ctx")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user