diff --git a/internal/engine/nats/nats.go b/internal/engine/nats/nats.go index 6bab124d74..cc09ebb42f 100644 --- a/internal/engine/nats/nats.go +++ b/internal/engine/nats/nats.go @@ -108,6 +108,10 @@ func (n *NatsEngine) Type() string { } func (n *NatsEngine) PublishTask(subject string, payload []byte) error { + if n.jetStream == nil { + return errors.New("NATS jetstream is nil, engine not properly initialized") + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -120,6 +124,10 @@ func (n *NatsEngine) PublishTask(subject string, payload []byte) error { } func (n *NatsEngine) ShowMessageQueue() (map[string]string, error) { + if n.jetStream == nil || n.stream == nil { + return nil, errors.New("NATS jetstream/stream is nil, engine not properly initialized") + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() accountInfo, err := n.jetStream.AccountInfo(ctx) @@ -226,6 +234,10 @@ func (n *NatsEngine) InitConsumer(subject string) error { return nil } func (n *NatsEngine) GetMessages(messageCount int) ([]common.TaskHandle, error) { + if n.consumer == nil { + return nil, errors.New("NATS consumer is nil, engine not properly initialized") + } + resultMessages := make([]common.TaskHandle, 0) messages, err := n.consumer.Fetch(messageCount, jetstream.FetchMaxWait(1*time.Second)) if err != nil { @@ -238,6 +250,9 @@ func (n *NatsEngine) GetMessages(messageCount int) ([]common.TaskHandle, error) } func (n *NatsEngine) CheckStatus() string { + if n.nc == nil { + return "NATS connection is nil, engine not properly initialized" + } n.nc.Stats() return n.nc.Status().String() } diff --git a/internal/engine/nats/nats_uninitialized_test.go b/internal/engine/nats/nats_uninitialized_test.go new file mode 100644 index 0000000000..2aade2f5a3 --- /dev/null +++ b/internal/engine/nats/nats_uninitialized_test.go @@ -0,0 +1,33 @@ +package nats + +import ( + "strings" + "testing" +) + +// An engine whose Init failed (or was never called) must return errors from +// its public methods instead of panicking on nil fields. Regression test for +// the nil-deref panic at PublishTask: the server bootstrap only logs message +// queue init failures, so a half-initialized engine stays registered +// process-wide and every later task publish crashed the caller. +func TestUninitializedEngineMethodsReturnErrors(t *testing.T) { + // Nothing listens on this port; Init() is deliberately not called so the + // engine keeps its zero-value (nil) connection, jetstream and consumer. + e := NewNatsEngine("127.0.0.1", 1) + + if err := e.PublishTask("tasks.RAGFLOW", []byte("{}")); err == nil || !strings.Contains(err.Error(), "not properly initialized") { + t.Fatalf("PublishTask on uninitialized engine: err = %v, want 'not properly initialized'", err) + } + + if _, err := e.GetMessages(1); err == nil || !strings.Contains(err.Error(), "not properly initialized") { + t.Fatalf("GetMessages on uninitialized engine: err = %v, want 'not properly initialized'", err) + } + + if _, err := e.ShowMessageQueue(); err == nil || !strings.Contains(err.Error(), "not properly initialized") { + t.Fatalf("ShowMessageQueue on uninitialized engine: err = %v, want 'not properly initialized'", err) + } + + if status := e.CheckStatus(); !strings.Contains(status, "not properly initialized") { + t.Fatalf("CheckStatus on uninitialized engine = %q, want 'not properly initialized'", status) + } +} diff --git a/internal/service/memory_queue_task_test.go b/internal/service/memory_queue_task_test.go new file mode 100644 index 0000000000..771e84b176 --- /dev/null +++ b/internal/service/memory_queue_task_test.go @@ -0,0 +1,34 @@ +package service + +import ( + "strings" + "testing" + + "ragflow/internal/engine" + natsengine "ragflow/internal/engine/nats" +) + +// queueMemoryTask is the agent-canvas memory-save publish path. When the +// message queue engine is a NatsEngine whose Init failed at boot, publishing +// must surface a clean error (caught by the Message component's best-effort +// memory save) rather than panicking and failing the whole canvas run. +func TestQueueMemoryTaskUninitializedQueueReturnsError(t *testing.T) { + previous := engine.GetMessageQueueEngine() + engine.SetMessageQueueEngine(natsengine.NewNatsEngine("127.0.0.1", 1)) + t.Cleanup(func() { engine.SetMessageQueueEngine(previous) }) + + err := queueMemoryTask( + t.Context(), + "mem-1", + "tenant-1", + 1, + map[string]any{"id": "task-1", "task_type": "memory"}, + MemoryMessage{UserID: "u1", AgentID: "agent-1", SessionID: "s1", UserInput: "hi", AgentResponse: "hello"}, + ) + if err == nil { + t.Fatal("queueMemoryTask with uninitialized MQ engine: err = nil, want publish error") + } + if !strings.Contains(err.Error(), "not properly initialized") { + t.Fatalf("queueMemoryTask err = %v, want engine-not-initialized error", err) + } +}