diff --git a/internal/engine/global.go b/internal/engine/global.go index 8d9657b74d..32a6f0f6b1 100644 --- a/internal/engine/global.go +++ b/internal/engine/global.go @@ -88,6 +88,13 @@ func GetMessageQueueEngine() MessageQueue { return messageQueueEngine } +// SetMessageQueueEngine installs the global message-queue engine. It exists +// primarily as a test seam so callers can drive Start() without a real server +// config; production code uses InitMessageQueueEngine. +func SetMessageQueueEngine(mq MessageQueue) { + messageQueueEngine = mq +} + func InitMessageQueueEngine(messageQueueType string) error { config := server.GetConfig() switch messageQueueType { diff --git a/internal/entity/knowledge_compile_doc.go b/internal/entity/knowledge_compile_doc.go index 80af64c9ab..5ac1c205e6 100644 --- a/internal/entity/knowledge_compile_doc.go +++ b/internal/entity/knowledge_compile_doc.go @@ -30,10 +30,14 @@ import "time" // out-of-order / tombstone guards as the broker-based design without re-reading // the queue. type KnowledgeCompileDoc struct { - DatasetID string `gorm:"primaryKey;column:dataset_id;size:64" json:"dataset_id"` - TenantID string `gorm:"column:tenant_id;size:64;not null;default:''" json:"tenant_id"` - BacklogDocIDs string `gorm:"column:backlog_doc_ids;type:text;not null;default:'[]'" json:"backlog_doc_ids"` - InflightDocIDs string `gorm:"column:inflight_doc_ids;type:text;not null;default:'[]'" json:"inflight_doc_ids"` + DatasetID string `gorm:"primaryKey;column:dataset_id;size:64" json:"dataset_id"` + TenantID string `gorm:"column:tenant_id;size:64;not null;default:''" json:"tenant_id"` + // The *_doc_ids columns store a JSON array as TEXT. No DDL default is set: + // MySQL (8.0.13+) rejects a literal DEFAULT on TEXT/BLOB columns (Error + // 1101), and the application always writes "[]" explicitly on insert/update + // (scheduler.go FirstOrCreate / release paths), so the default is redundant. + BacklogDocIDs string `gorm:"column:backlog_doc_ids;type:text;not null" json:"backlog_doc_ids"` + InflightDocIDs string `gorm:"column:inflight_doc_ids;type:text;not null" json:"inflight_doc_ids"` ClaimOwner string `gorm:"column:claim_owner;size:64;not null;default:''" json:"claim_owner"` ClaimToken string `gorm:"column:claim_token;size:64;not null;default:''" json:"claim_token"` ClaimExpiresAt *time.Time `gorm:"column:claim_expires_at;default:null" json:"claim_expires_at"` diff --git a/internal/ingestion/service/ingestion_service.go b/internal/ingestion/service/ingestion_service.go index d4b2b4bdeb..877c97186b 100644 --- a/internal/ingestion/service/ingestion_service.go +++ b/internal/ingestion/service/ingestion_service.go @@ -64,10 +64,11 @@ type Ingestor struct { ShutdownCh chan struct{} // Worker pool - taskChan chan *taskpkg.TaskContext - workerWg sync.WaitGroup - startOnce sync.Once - stopOnce sync.Once // guards close(ShutdownCh) against double-close on repeated Stop + taskChan chan *taskpkg.TaskContext + workerWg sync.WaitGroup + startOnce sync.Once + workerOnce sync.Once // guards startWorkerPool; must NOT be startOnce (Start wraps start() in startOnce, and start() calls startWorkerPool -> re-entry deadlock) + stopOnce sync.Once // guards close(ShutdownCh) against double-close on repeated Stop ingestionTaskSvc *servicepkg.IngestionTaskService docState *docStateUpdater @@ -382,7 +383,7 @@ func (e *Ingestor) processMessage(handle common.TaskHandle) { } func (e *Ingestor) startWorkerPool() { - e.startOnce.Do(func() { + e.workerOnce.Do(func() { for i := int32(0); i < e.maxConcurrency; i++ { e.workerWg.Add(1) go e.workerLoop(i) diff --git a/internal/ingestion/service/ingestor_lifecycle_test.go b/internal/ingestion/service/ingestor_lifecycle_test.go index 3c4ce2cab5..cb95f882a4 100644 --- a/internal/ingestion/service/ingestor_lifecycle_test.go +++ b/internal/ingestion/service/ingestor_lifecycle_test.go @@ -22,6 +22,7 @@ import ( "time" "ragflow/internal/common" + "ragflow/internal/engine" "ragflow/internal/entity" taskpkg "ragflow/internal/ingestion/task" "ragflow/internal/ingestion/testutil" @@ -200,3 +201,49 @@ func TestPollCancel_ExitsWhenDoneClosed(t *testing.T) { close(released) // cleanup } + +// TestStart_FullPathReturnsAndStartsWorkers is a regression test for the +// sync.Once re-entrancy deadlock. Before the fix, Start() wrapped the whole +// startup (start()) in e.startOnce.Do, but start() also called startWorkerPool() +// which nested the SAME startOnce. sync.Once.Do blocks forever when re-entered +// from inside its own callback, so Start() hung after InitConsumer succeeded: +// no worker pool, no consumeLoop, and ingestion tasks were never consumed. +// +// The test drives the real Start() path (start -> startWorkerPool -> consumeLoop) +// against an embedded NATS server and asserts Start() returns within a deadline +// and that workers are actually up. +func TestStart_FullPathReturnsAndStartsWorkers(t *testing.T) { + // SetMessageQueueEngine mutates process-global state; restore the previous + // engine so later tests don't inherit a closed embedded NATS server. + previousEngine := engine.GetMessageQueueEngine() + engine.SetMessageQueueEngine(testutil.SetupNatsEngine(t)) + t.Cleanup(func() { engine.SetMessageQueueEngine(previousEngine) }) + + const concurrency int32 = 2 + ing := NewIngestor("test-start-fullpath", concurrency, nil) + t.Cleanup(func() { ing.Stop(context.Background()) }) + + done := make(chan error, 1) + go func() { + done <- ing.Start() + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Start() returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Start() did not return within 10s; sync.Once re-entrancy deadlock likely") + } + + // Start() launches workers asynchronously and returns immediately; poll + // briefly so we don't observe zero before a worker enters workerLoop. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) && ing.activeWorkers.Load() <= 0 { + time.Sleep(time.Millisecond) + } + if got := ing.activeWorkers.Load(); got <= 0 { + t.Fatalf("expected activeWorkers > 0 after Start(), got %d", got) + } +}