From 3ef6e45e438c9faeed00d7789cf8edc50cd41325 Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Tue, 11 Aug 2026 19:36:35 +0800 Subject: [PATCH] fix: return error when chat-channel start has error (#18068) --- cmd/ragflow_server.go | 5 ++- internal/channels/bootstrap.go | 60 ++++++++++++++++++++-------------- 2 files changed, 40 insertions(+), 25 deletions(-) diff --git a/cmd/ragflow_server.go b/cmd/ragflow_server.go index 6d7c5cd204..75898b31f9 100644 --- a/cmd/ragflow_server.go +++ b/cmd/ragflow_server.go @@ -919,7 +919,10 @@ func startServer(ctx context.Context) { // Setup routes r.Setup(ginEngine) - channels.Start(ctx) + _, err := channels.Start(ctx) + if err != nil { + common.Fatal("Fail to start chat-channel", zap.Error(err)) + } apiServerConfig := globalConfig.GetAPIServerConfig() diff --git a/internal/channels/bootstrap.go b/internal/channels/bootstrap.go index 7179233140..24dca10fe3 100644 --- a/internal/channels/bootstrap.go +++ b/internal/channels/bootstrap.go @@ -39,12 +39,12 @@ const ( ) type runningChannel struct { - channel core.Channel - fp string + channel core.Channel + fingerprint string } type failedChannel struct { - fp string + fingerprint string attempts int nextRetryAt time.Time } @@ -64,19 +64,25 @@ func NewRuntime() *Runtime { } // Start launches the chat-channel runtime reconciler in the background. -func Start(ctx context.Context) *Runtime { - rt := NewRuntime() +func Start(ctx context.Context) (*Runtime, error) { + channelRuntime := NewRuntime() service.SetChatChannelRuntimeProvider(whatsappRuntimeSnapshotMap) - go rt.Run(ctx) - return rt + + if err := channelRuntime.Reconcile(ctx); err != nil { + return channelRuntime, fmt.Errorf("failed to reconcile channel: %w", err) + } + + go channelRuntime.Run(ctx) + return channelRuntime, nil } // Run reconciles configured chat channels until the context is cancelled. func (r *Runtime) Run(ctx context.Context) { ticker := time.NewTicker(reconcileInterval) defer ticker.Stop() - defer r.stopAll(context.Background()) + defer r.stopAll(ctx) + // check channel's status periodically for { if err := r.Reconcile(ctx); err != nil { log.Printf("chat channel reconcile failed: %v", err) @@ -91,6 +97,7 @@ func (r *Runtime) Run(ctx context.Context) { // Reconcile starts, stops, and restarts channel instances to match the database state. func (r *Runtime) Reconcile(ctx context.Context) error { + var reconcileErr error desired, err := desiredChannels(ctx) if err != nil { return err @@ -100,21 +107,22 @@ func (r *Runtime) Reconcile(ctx context.Context) error { r.mu.Lock() for accountID, entry := range r.running { wanted, ok := desired[accountID] - if !ok || wanted.fp != entry.fp { + if !ok || wanted.fingerprint != entry.fingerprint { delete(r.running, accountID) toStop = append(toStop, entry.channel) } } + for accountID, failure := range r.failed { wanted, ok := desired[accountID] - if !ok || wanted.fp != failure.fp { + if !ok || wanted.fingerprint != failure.fingerprint { delete(r.failed, accountID) } } r.mu.Unlock() for _, ch := range toStop { - stopChannel(context.Background(), ch) + stopChannel(ctx, ch) } activeWhatsApp := false @@ -126,6 +134,7 @@ func (r *Runtime) Reconcile(ctx context.Context) error { } if err = syncWhatsAppGateway(ctx, activeWhatsApp); err != nil && activeWhatsApp { log.Printf("failed to sync WhatsApp gateway: %v", err) + reconcileErr = fmt.Errorf("sync WhatsApp gateway: %w", err) } now := time.Now() @@ -133,32 +142,35 @@ func (r *Runtime) Reconcile(ctx context.Context) error { r.mu.Lock() _, isRunning := r.running[accountID] failure, failed := r.failed[accountID] - retryPending := failed && failure.fp == wanted.fp && now.Before(failure.nextRetryAt) + retryPending := failed && failure.fingerprint == wanted.fingerprint && now.Before(failure.nextRetryAt) r.mu.Unlock() if isRunning || retryPending { continue } if err = r.startChannel(ctx, accountID, wanted); err != nil { log.Printf("failed to start chat channel %s (%s): %v", accountID, wanted.channel, err) - r.recordStartFailure(accountID, wanted.fp, now) + r.recordStartFailure(accountID, wanted.fingerprint, now) + if reconcileErr == nil { + reconcileErr = fmt.Errorf("failed to start chat channel %s: %w", accountID, err) + } continue } r.clearStartFailure(accountID) } - return nil + return reconcileErr } // recordStartFailure saves the next retry window for a failed channel start. -func (r *Runtime) recordStartFailure(accountID string, fp string, now time.Time) { +func (r *Runtime) recordStartFailure(accountID string, fingerprint string, now time.Time) { r.mu.Lock() defer r.mu.Unlock() attempts := 1 - if failure, ok := r.failed[accountID]; ok && failure.fp == fp { + if failure, ok := r.failed[accountID]; ok && failure.fingerprint == fingerprint { attempts = failure.attempts + 1 } r.failed[accountID] = failedChannel{ - fp: fp, + fingerprint: fingerprint, attempts: attempts, nextRetryAt: now.Add(startRetryDelay(attempts)), } @@ -187,9 +199,9 @@ func startRetryDelay(attempts int) time.Duration { } type desiredChannel struct { - channel string - credential map[string]any - fp string + channel string + credential map[string]any + fingerprint string } // desiredChannels loads enabled chat-channel rows and reduces them to runtime configuration. @@ -202,9 +214,9 @@ func desiredChannels(ctx context.Context) (map[string]desiredChannel, error) { for _, row := range rows { credential := credentialFromConfig(row.Config) out[row.ID] = desiredChannel{ - channel: row.Channel, - credential: credential, - fp: fingerprint(row.Channel, credential), + channel: row.Channel, + credential: credential, + fingerprint: fingerprint(row.Channel, credential), } } return out, nil @@ -268,7 +280,7 @@ func (r *Runtime) startChannel(ctx context.Context, accountID string, wanted des return err } r.mu.Lock() - r.running[accountID] = runningChannel{channel: ch, fp: wanted.fp} + r.running[accountID] = runningChannel{channel: ch, fingerprint: wanted.fingerprint} r.mu.Unlock() log.Printf("started chat channel %s:%s", ch.ChannelID(), accountID) return nil